TestAddLpEncodeToJndn
说明：本文所说的Tag即 NDNLPv2 协议中的 LpHeaderField
本文涉及的测试代码源码：TestAddLpEncodeToJndn
根据 NDN新增tag的方案 提供的新增Tag方案，修改ndn-cxx 和 NFD 之后支持若干个新的Tag，本项目用于验证根据 jndn库调整支持setCongestionMark 和 在jndn中添加一种新的LpHeaderField支持 对 jndn 库进行对应修改之后可以支持对新增tag的读取和写入。
1. 场景1
首先针对 jndn 库对Tag写入功能进行测试，期望使用 jndn 库实现一个 Consumer，用 ndn-cxx 库实现一个Producer，接着 Consumer 在发送 Interest 时，在其中携带若干个Tag，期望在 Producer 测能够收到对应的 Tag ，并且可以正确的解析对应 Tag 的值。
- 测试实验示意图如下：
-
Consumer 端代码：
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 端代码：
#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 的值。
- 测试实验示意图如下：
-
Consumer 端代码：
#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 端代码：
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); } } }