From b61e3e217473e59a4365eea9f88d58b91fd88c0b Mon Sep 17 00:00:00 2001 From: Davide Pesavento Date: Fri, 11 Nov 2022 14:38:52 -0500 Subject: [PATCH] putchunks: use ndn-cxx's Segmenter Change-Id: I2a2bdac2983cce45c6e6d8f5b41a5609b4a98c48 --- tools/chunks/putchunks/producer.cpp | 70 +++++++++-------------------- tools/chunks/putchunks/producer.hpp | 24 +++------- 2 files changed, 26 insertions(+), 68 deletions(-) diff --git a/tools/chunks/putchunks/producer.cpp b/tools/chunks/putchunks/producer.cpp index 24565c0..6b305cd 100644 --- a/tools/chunks/putchunks/producer.cpp +++ b/tools/chunks/putchunks/producer.cpp @@ -31,6 +31,7 @@ #include "producer.hpp" #include +#include namespace ndn::chunks { @@ -49,12 +50,13 @@ Producer::Producer(const Name& prefix, Face& face, KeyChain& keyChain, std::istr m_versionedPrefix = Name(m_prefix).appendVersion(); } - populateStore(is); + if (!m_options.isQuiet) { + std::cerr << "Loading input ...\n"; + } + util::Segmenter segmenter(m_keyChain, m_options.signingInfo); + m_store = segmenter.segment(is, m_versionedPrefix, m_options.maxSegmentSize, m_options.freshnessPeriod); - if (m_options.wantShowVersion) - std::cout << m_versionedPrefix[-1] << "\n"; - - // register m_prefix without interest handler + // register m_prefix without Interest handler m_face.registerPrefix(m_prefix, nullptr, [this] (const Name& prefix, const auto& reason) { std::cerr << "ERROR: Failed to register prefix '" << prefix << "' (" << reason << ")\n"; m_face.shutdown(); @@ -76,8 +78,13 @@ Producer::Producer(const Name& prefix, Face& face, KeyChain& keyChain, std::istr processDiscoveryInterest(interest); }); - if (!m_options.isQuiet) - std::cerr << "Data published with name: " << m_versionedPrefix << "\n"; + if (m_options.wantShowVersion) { + std::cout << m_versionedPrefix[-1] << "\n"; + } + if (!m_options.isQuiet) { + std::cerr << "Published " << m_store.size() << " Data packet" << (m_store.size() > 1 ? "s" : "") + << " with prefix " << m_versionedPrefix << "\n"; + } } void @@ -93,8 +100,9 @@ Producer::processDiscoveryInterest(const Interest& interest) std::cerr << "Discovery Interest: " << interest << "\n"; if (!interest.getCanBePrefix()) { - if (m_options.isVerbose) + if (m_options.isVerbose) { std::cerr << "Discovery Interest lacks CanBePrefix, sending Nack\n"; + } m_face.put(lp::Nack(interest)); return; } @@ -103,7 +111,7 @@ Producer::processDiscoveryInterest(const Interest& interest) mobject.setVersionedName(m_versionedPrefix); // make a metadata packet based on the received discovery Interest name - Data mdata(mobject.makeData(interest.getName(), m_keyChain, m_options.signingInfo)); + auto mdata = mobject.makeData(interest.getName(), m_keyChain, m_options.signingInfo); if (m_options.isVerbose) std::cerr << "Sending metadata: " << mdata << "\n"; @@ -135,53 +143,17 @@ Producer::processSegmentInterest(const Interest& interest) } if (data != nullptr) { - if (m_options.isVerbose) + if (m_options.isVerbose) { std::cerr << "Data: " << *data << "\n"; - + } m_face.put(*data); } else { - if (m_options.isVerbose) + if (m_options.isVerbose) { std::cerr << "Interest cannot be satisfied, sending Nack\n"; + } m_face.put(lp::Nack(interest)); } } -void -Producer::populateStore(std::istream& is) -{ - BOOST_ASSERT(m_store.empty()); - - if (!m_options.isQuiet) - std::cerr << "Loading input ...\n"; - - std::vector buffer(m_options.maxSegmentSize); - while (is.good()) { - is.read(reinterpret_cast(buffer.data()), buffer.size()); - const auto nCharsRead = is.gcount(); - - if (nCharsRead > 0) { - auto data = make_shared(Name(m_versionedPrefix).appendSegment(m_store.size())); - data->setFreshnessPeriod(m_options.freshnessPeriod); - data->setContent(make_span(buffer).first(nCharsRead)); - m_store.push_back(data); - } - } - - if (m_store.empty()) { - auto data = make_shared(Name(m_versionedPrefix).appendSegment(0)); - data->setFreshnessPeriod(m_options.freshnessPeriod); - m_store.push_back(data); - } - - auto finalBlockId = name::Component::fromSegment(m_store.size() - 1); - for (const auto& data : m_store) { - data->setFinalBlock(finalBlockId); - m_keyChain.sign(*data, m_options.signingInfo); - } - - if (!m_options.isQuiet) - std::cerr << "Created " << m_store.size() << " chunks for prefix " << m_prefix << "\n"; -} - } // namespace ndn::chunks diff --git a/tools/chunks/putchunks/producer.hpp b/tools/chunks/putchunks/producer.hpp index 7000bd7..afacabd 100644 --- a/tools/chunks/putchunks/producer.hpp +++ b/tools/chunks/putchunks/producer.hpp @@ -35,7 +35,7 @@ namespace ndn::chunks { /** - * @brief Segmented & versioned data publisher + * @brief Segmented & versioned data publisher. * * Packetizes and publishes data from an input stream as `/prefix//`. * Unless another value is provided, the current time is used as the version number. @@ -54,9 +54,8 @@ public: bool wantShowVersion = false; }; -public: /** - * @brief Create the producer + * @brief Create the producer. * @param prefix prefix used to publish data; if the last component is not a valid * version number, the current system time is used as version number. */ @@ -64,33 +63,20 @@ public: const Options& opts); /** - * @brief Run the producer + * @brief Run the producer. */ void run(); private: /** - * @brief Split the input stream in data packets and save them to the store - * - * Create data packets reading all the characters from the input stream until EOF or an - * error occurs. Each data packet has a maximum payload size of `m_options.maxSegmentSize` - * bytes and is stored in the vector `m_store`. An empty data packet is created and stored - * if the input stream is empty. - * - * @return Number of data packets contained in the store after the operation - */ - void - populateStore(std::istream& is); - - /** - * @brief Respond with a metadata packet containing the versioned content name + * @brief Respond with a metadata packet containing the versioned content name. */ void processDiscoveryInterest(const Interest& interest); /** - * @brief Respond with the requested segment of content + * @brief Respond with the requested segment of content. */ void processSegmentInterest(const Interest& interest);