putchunks: use ndn-cxx's Segmenter

Change-Id: I2a2bdac2983cce45c6e6d8f5b41a5609b4a98c48
This commit is contained in:
Davide Pesavento committed 2022-11-11 14:38:52 -05:00
1 parent 3d7b03372b
commit b61e3e2174
2 files changed
+26 -68

No files matched your search

+21 -49
View File
@@ -31,6 +31,7 @@
#include "producer.hpp"
#include <ndn-cxx/metadata-object.hpp>
#include <ndn-cxx/util/segmenter.hpp>
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<uint8_t> buffer(m_options.maxSegmentSize);
while (is.good()) {
is.read(reinterpret_cast<char*>(buffer.data()), buffer.size());
const auto nCharsRead = is.gcount();
if (nCharsRead > 0) {
auto data = make_shared<Data>(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<Data>(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
+5 -19
View File
@@ -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/<version>/<segment number>`.
* 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);