SunnyQjm/TestAddLpEncodeToJndn
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

update: README

Browse Source
This commit is contained in:
SunnyQjm
2020-07-24 19:11:56 +08:00
parent 54972f8cb1
commit 94c341fe55
1 changed files with 364 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+364
View File
@@ -0,0 +1,364 @@
# TestAddLpEncodeToJndn
> 说明：本文所说的Tag即 NDNLPv2 协议中的 LpHeaderField
>
> 本文涉及的测试代码源码：[TestAddLpEncodeToJndn](https://gitea.qjm253.cn/SunnyQjm/TestAddLpEncodeToJndn)
根据 [NDN新增tag的方案](https://blog.csdn.net/qq_33215972/article/details/105635230) 提供的新增Tag方案，修改`ndn-cxx` 和 `NFD` 之后支持若干个新的Tag，本项目用于验证根据 [jndn库调整支持setCongestionMark](https://gitea.qjm253.cn/PKUSZ-future-network-lab/jndn/src/branch/MIN/docs/modify-jndn-to-support-setCongestionMark.md) 和 [在jndn中添加一种新的LpHeaderField支持](https://gitea.qjm253.cn/PKUSZ-future-network-lab/jndn/src/branch/MIN/docs/add-new-tag-to-jndn) 对 `jndn` 库进行对应修改之后可以支持对新增tag的读取和写入。
### 1. 场景1
首先针对 `jndn` 库对Tag写入功能进行测试，期望使用 `jndn` 库实现一个 Consumer，用 `ndn-cxx` 库实现一个Producer，接着 Consumer 在发送 `Interest` 时，在其中携带若干个Tag，期望在 Producer 测能够收到对应的 Tag ，并且可以正确的解析对应 Tag 的值。
- 测试实验示意图如下：
![TestLpEncodeInJndn](README.assets/TestLpEncodeInJndn.png)
- Consumer 端代码：
```java
public class Consumer {
public static void main(String[] args) throws IOException, EncodingException, InterruptedException {
WireFormat.setDefaultWireFormat(new Tlv0_3WireFormat());
Name interestName = new Name("/test").append("hello" + System.currentTimeMillis());
Interest interest = new Interest(interestName);
interest.setCongestionMark(1);
interest.setPktType(PktType.PKT_TYPE_ID_PKT);
interest.setSrcAddr(new Name("/testSrcAddr"));
interest.setDstAddr(new Name("/testDstAddr"));
interest.setBackupName(new Name("/testBackupName"));
interest.setCachePolicy(CachePolicy.CACHE_POLICY_TYPE_NO_CACHE);
System.out.println(interest.getSrcAddr().toUri());
final int[] dataCallbackCount = new int[]{0};
final int[] timeoutCallbackCount = new int[]{0};
final Data[] receivedData = new Data[1];
Face face = new Face();
face.expressInterest(interest, (interest1, data) -> {
dataCallbackCount[0]++;
receivedData[0] = data;
}, interest1 -> {
timeoutCallbackCount[0]++;
});
long time = 10000;
boolean done = false;
long startTime = System.currentTimeMillis();
while (System.currentTimeMillis() - startTime < time) {
face.processEvents();public class Producer {
public static void main(String[] args) throws IOException, SecurityException, EncodingException, InterruptedException, PibImpl.Error, KeyChain.Error {
WireFormat.setDefaultWireFormat(new Tlv0_3WireFormat());
Face face = new Face();
KeyChain keyChain = new KeyChain();
face.setCommandSigningInfo(keyChain, keyChain.getDefaultCertificateName());
face.registerPrefix(new Name("/test"), (name, interest, face1, l, interestFilter) -> { // onInterest
assert 1L == interest.getCongestionMark();
assert PktType.PKT_TYPE_ID_PKT == interest.getPktType();
assert "/testSrcAddr".equals(interest.getSrcAddr().toUri());
assert "/testDstAddr".equals(interest.getDstAddr().toUri());
assert "/testBackupName".equals(interest.getBackupName().toUri());
assert CachePolicy.CACHE_POLICY_TYPE_NO_CACHE == interest.getCachePolicy();
Data data = new Data(interest.getName());
data.setContent(new Blob("hello"));
data.setCongestionMark(1);
data.setPktType(PktType.PKT_TYPE_ID_PKT);
data.setSrcAddr(new Name("/testSrcAddr"));
data.setDstAddr(new Name("/testDstAddr"));
data.setBackupName(new Name("/testBackupName"));
data.setCachePolicy(CachePolicy.CACHE_POLICY_TYPE_NO_CACHE);
try {
keyChain.sign(data);
} catch (SecurityException | TpmBackEnd.Error | PibImpl.Error | KeyChain.Error e) {
e.printStackTrace();
}
try {
face.putData(data);
} catch (IOException e) {
e.printStackTrace();
}
}, name -> { // onRegisterFailed
System.out.println("onRegisterFailed: " + name.toUri());
});
while (true) {
face.processEvents();
Thread.sleep(10);
}
}
}
if (dataCallbackCount[0] > 0 || timeoutCallbackCount[0] > 0) {
break;
}
Thread.sleep(10);
}
assert 1L == receivedData[0].getCongestionMark();
assert PktType.PKT_TYPE_ID_PKT == receivedData[0].getPktType();
assert "/testSrcAddr".equals(receivedData[0].getSrcAddr().toUri());
assert "/testDstAddr".equals(receivedData[0].getDstAddr().toUri());
assert "/testBackupName".equals(receivedData[0].getBackupName().toUri());
assert CachePolicy.CACHE_POLICY_TYPE_NO_CACHE == receivedData[0].getCachePolicy();
}
}
```
- Producer 端代码：
```cpp
#include <ndn-cxx/face.hpp>
#include <ndn-cxx/security/key-chain.hpp>
#include <ndn-cxx/security/signing-helpers.hpp>
#include <ndn-cxx/lp/tags.hpp>
#include <ndn-cxx/lp/lp-headers.hpp>
#include <iostream>
using namespace std;
// Enclosing code in ndn simplifies coding (can also use `using namespace ndn`)
namespace ndn {
// Additional nested namespaces should be used to prevent/limit name conflicts
namespace examples {
class Producer {
public:
Producer() {}
void
run() {
cout << "before setInterestFilter" << endl;
// 向NFD注册一个名称前缀 /example/testApp
// 如果成功，NFD会在接收到以这个前缀开头的兴趣包转发给本程序，Producer::onInterest回调会被调用
m_face.setInterestFilter("/test",
bind(&Producer::onInterest, this, _1, _2),
nullptr, // RegisterPrefixSuccessCallback is optional
bind(&Producer::onRegisterFailed, this, _1, _2));
cout << "before processEvents" << endl;
// 处理NDN事件，Producer::onInterest回调是有这个函数里面调用的
// 意味着Producer::onInterest回调会在调用 face.processEvents 的线程运行
// 这个函数是一个阻塞函数，调用这个函数会期望处理所有相关的事件，如果向上面那样注册了前缀监听，则会一直处理事件
m_face.processEvents();
cout << "after processEvents" << endl;
}
private:
void
onInterest(const InterestFilter &, const Interest &interest) {
std::cout << ">> I: " << interest << std::endl;
cout << "CongestionMark: " << interest.getTag<lp::CongestionMarkTag>()->get() << endl;
assert(interest.getTag<lp::CongestionMarkTag>()->get() == 1);
cout << "PktType: " << interest.getTag<lp::PktTypeTag>()->get() << endl;
assert(interest.getTag<lp::PktTypeTag>()->get() == 2);
cout << "SrcAddr: " << interest.getTag<lp::SrcAddrTag>()->get().getMyPrefix()->toUri() << endl;
assert(interest.getTag<lp::SrcAddrTag>()->get().getMyPrefix()->toUri() == "/testSrcAddr");
cout << "DstAddr: " << interest.getTag<lp::DstAddrTag>()->get().getMyPrefix()->toUri() << endl;
assert(interest.getTag<lp::DstAddrTag>()->get().getMyPrefix()->toUri() == "/testDstAddr");
cout << "BackupName: " << interest.getTag<lp::BackupNameTag>()->get().getMyPrefix()->toUri() << endl;
assert(interest.getTag<lp::BackupNameTag>()->get().getMyPrefix()->toUri() == "/testBackupName");
cout << "CachePolicy: " << interest.getTag<lp::CachePolicyTag>()->get().getPolicy() << endl;
assert(interest.getTag<lp::CachePolicyTag>()->get().getPolicy() == lp::CachePolicyType::NO_CACHE);
static const std::string content("Hello, world!");
// Create Data packet
auto data = make_shared<Data>(interest.getName());
data->setFreshnessPeriod(10_s);
data->setContent(reinterpret_cast<const uint8_t *>(content.data()), content.size());
data->setTag(make_shared<lp::CongestionMarkTag>(1));
data->setTag(make_shared<lp::PktTypeTag>(2));
data->setTag(make_shared<lp::SrcAddrTag>(lp::SrcAddrHeader(Name("/testSrcAddr"))));
data->setTag(make_shared<lp::DstAddrTag>(lp::DstAddrHeader(Name("/testDstAddr"))));
data->setTag(make_shared<lp::BackupNameTag>(lp::BackupNameHeader(Name("/testBackupName"))));
auto cachePolicy = lp::CachePolicy();
cachePolicy.setPolicy(lp::CachePolicyType::NO_CACHE);
data->setTag(make_shared<lp::CachePolicyTag>(cachePolicy));
// Sign Data packet with default identity
m_keyChain.sign(*data);
std::cout << "<< D: " << *data << std::endl;
m_face.put(*data);
}
void
onRegisterFailed(const Name &prefix, const std::string &reason) {
std::cerr << "ERROR: Failed to register prefix '" << prefix
<< "' with the local forwarder (" << reason << ")" << std::endl;
m_face.shutdown();
}
private:
Face m_face;
KeyChain m_keyChain;
};
} // namespace examples
} // namespace ndn
int
main(int argc, char **argv) {
try {
ndn::examples::Producer producer;
producer.run();
return 0;
}
catch (const std::exception &e) {
std::cerr << "ERROR: " << e.what() << std::endl;
return 1;
}
}
```
### 2. 场景2
接着针对 `jndn` 库对Tag解析功能进行测试，期望使用 `ndn-cxx` 库实现一个 Consumer，用 `jndn` 库实现一个Producer，接着 Consumer 在发送 `Interest` 时，在其中携带若干个Tag，期望在 Producer 测能够收到对应的 Tag ，并且可以正确的解析对应 Tag 的值。
- 测试实验示意图如下：
![TestLpEncodeInJndn1](README.assets/TestLpEncodeInJndn1-1595588289562.png)
- Consumer 端代码：
```cpp
#include <ndn-cxx/face.hpp>
#include <ndn-cxx/lp/tags.hpp>
#include <ndn-cxx/lp/lp-headers.hpp>
#include <string>
#include <iostream>
// Enclosing code in ndn simplifies coding (can also use `using namespace ndn`)
namespace ndn {
// Additional nested namespaces can be used to prevent/limit name conflicts
namespace examples {
class Consumer : noncopyable {
public:
void
run() {
Interest interest(Name("/test/hello/" + std::to_string(1)));
// 如果需要传递参数，可以通过一下方式传递
std::string param = "{\"code\": 1, \"msg\": \"balabala\"}";
interest.setApplicationParameters(reinterpret_cast<const uint8_t *>(param.data()), param.size());
// 设置兴趣包的生存期，超过这个生存期没有收到对应的数据包，兴趣包就会触发超时
interest.setInterestLifetime(2_s); // 2 seconds
// 设置如果强制刷新，则不会命中途中路由器的缓存
interest.setMustBeFresh(true);
interest.setTag(make_shared<lp::CongestionMarkTag>(1));
interest.setTag(make_shared<lp::PktTypeTag>(2));
auto srcTag = make_shared<lp::SrcAddrTag>(lp::SrcAddrHeader(Name("/testSrcAddr")));
interest.setTag(srcTag);
srcTag->get().getMyPrefix()->wireEncode();
interest.setTag(make_shared<lp::DstAddrTag>(lp::DstAddrHeader(Name("/testDstAddr"))));
interest.setTag(make_shared<lp::BackupNameTag>(lp::BackupNameHeader(Name("/testBackupName"))));
auto cachePolicy = lp::CachePolicy();
cachePolicy.setPolicy(lp::CachePolicyType::NO_CACHE);
interest.setTag(make_shared<lp::CachePolicyTag>(cachePolicy));
// 发送兴趣包
m_face.expressInterest(interest,
bind(&Consumer::onData, this, _1, _2),
bind(&Consumer::onNack, this, _1, _2),
bind(&Consumer::onTimeout, this, _1));
std::cout << "Sending " << interest << std::endl;
// 需要调用processEvents处理NDN事件，这样才能处理收到的数据包，要不然程序就直接退出了
// 这个操作会一直阻塞，直到处理完所有的NDN事件
// processEvents will block until the requested data received or timeout occurs
m_face.processEvents();
}
private:
void
onData(const Interest &interest, const Data &data) {
std::cout << data << std::endl;
}
void
onNack(const Interest &interest, const lp::Nack &nack) {
std::cout << "received Nack with reason " << nack.getReason()
<< " for interest " << interest << std::endl;
}
void
onTimeout(const Interest &interest) {
std::cout << "Timeout " << interest << std::endl;
}
private:
Face m_face;
};
} // namespace examples
} // namespace ndn
int
main(int argc, char **argv) {
ndn::examples::Consumer consumer;
try {
consumer.run();
}
catch (const std::exception &e) {
std::cerr << "ERROR: " << e.what() << std::endl;
}
return 0;
}
```
- Producer 端代码：
```java
public class Producer {
public static void main(String[] args) throws IOException, SecurityException, EncodingException, InterruptedException, PibImpl.Error, KeyChain.Error {
WireFormat.setDefaultWireFormat(new Tlv0_3WireFormat());
Face face = new Face();
KeyChain keyChain = new KeyChain();
face.setCommandSigningInfo(keyChain, keyChain.getDefaultCertificateName());
face.registerPrefix(new Name("/test"), (name, interest, face1, l, interestFilter) -> { // onInterest
assert 1L == interest.getCongestionMark();
assert PktType.PKT_TYPE_ID_PKT == interest.getPktType();
assert "/testSrcAddr".equals(interest.getSrcAddr().toUri());
assert "/testDstAddr".equals(interest.getDstAddr().toUri());
assert "/testBackupName".equals(interest.getBackupName().toUri());
assert CachePolicy.CACHE_POLICY_TYPE_NO_CACHE == interest.getCachePolicy();
Data data = new Data(interest.getName());
data.setContent(new Blob("hello"));
data.setCongestionMark(1);
data.setPktType(PktType.PKT_TYPE_ID_PKT);
data.setSrcAddr(new Name("/testSrcAddr"));
data.setDstAddr(new Name("/testDstAddr"));
data.setBackupName(new Name("/testBackupName"));
data.setCachePolicy(CachePolicy.CACHE_POLICY_TYPE_NO_CACHE);
try {
keyChain.sign(data);
} catch (SecurityException | TpmBackEnd.Error | PibImpl.Error | KeyChain.Error e) {
e.printStackTrace();
}
try {
face.putData(data);
} catch (IOException e) {
e.printStackTrace();
}
}, name -> { // onRegisterFailed
System.out.println("onRegisterFailed: " + name.toUri());
});
while (true) {
face.processEvents();
Thread.sleep(10);
}
}
}
```