23 #include <ndn-cxx/util/logger.hpp> 24 #include <ndn-cxx/security/validator-null.hpp> 26 #include <boost/algorithm/string.hpp> 37 double false_positive = 0.001,
38 ndn::time::milliseconds helloInterestLifetime,
39 ndn::time::milliseconds syncInterestLifetime)
41 , m_scheduler(m_face.getIoService())
42 , m_syncPrefix(syncPrefix)
43 , m_helloInterestPrefix(ndn::Name(m_syncPrefix).append(
"hello"))
44 , m_syncInterestPrefix(ndn::Name(m_syncPrefix).append(
"sync"))
45 , m_syncDataContentType(ndn::tlv::ContentType_Blob)
46 , m_onReceiveHelloData(onReceiveHelloData)
47 , m_onUpdate(onUpdate)
48 , m_bloomFilter(count, false_positive)
49 , m_helloInterestLifetime(helloInterestLifetime)
50 , m_syncInterestLifetime(syncInterestLifetime)
51 , m_rng(ndn::random::getRandomNumberEngine())
52 , m_rangeUniformRandom(100, 500)
59 auto it = m_prefixes.insert(std::pair<ndn::Name, uint64_t>(prefix, 0));
63 m_subscriptionList.insert(prefix);
64 m_bloomFilter.
insert(prefix.toUri());
71 m_scheduler.cancelAllEvents();
74 m_syncFetcher->stop();
75 m_syncFetcher.reset();
79 m_helloFetcher->stop();
80 m_helloFetcher.reset();
87 ndn::Interest helloInterest(m_helloInterestPrefix);
88 NDN_LOG_DEBUG(
"Send Hello Interest " << helloInterest);
91 m_helloFetcher->stop();
94 using ndn::util::SegmentFetcher;
95 SegmentFetcher::Options options;
96 options.interestLifetime = m_helloInterestLifetime;
97 options.maxTimeout = m_helloInterestLifetime;
98 options.rttOptions.initialRto = m_syncInterestLifetime;
100 m_helloFetcher = SegmentFetcher::start(m_face, helloInterest,
101 ndn::security::v2::getAcceptAllValidator(), options);
103 m_helloFetcher->afterSegmentValidated.connect([
this] (
const ndn::Data& data) {
104 if (data.getFinalBlock()) {
105 m_helloDataName = data.getName().getPrefix(-2);
109 m_helloFetcher->onComplete.connect([
this] (
const ndn::ConstBufferPtr& bufferPtr) {
110 onHelloData(bufferPtr);
113 m_helloFetcher->onError.connect([
this] (uint32_t errorCode,
const std::string& msg) {
114 NDN_LOG_TRACE(
"Cannot fetch hello data, error: " << errorCode <<
" message: " << msg);
115 ndn::time::milliseconds after(m_rangeUniformRandom(m_rng));
116 NDN_LOG_TRACE(
"Scheduling after " << after);
122 Consumer::onHelloData(
const ndn::ConstBufferPtr& bufferPtr)
124 NDN_LOG_DEBUG(
"On Hello Data");
127 m_iblt = m_helloDataName.getSubName(m_helloDataName.size()-1, 1);
129 NDN_LOG_TRACE(
"m_iblt: " << std::hash<std::string>{}(m_iblt.toUri()));
131 State state{ndn::Block{bufferPtr}};
133 std::vector<MissingDataInfo> updates;
134 std::vector<ndn::Name> availableSubscriptions;
136 NDN_LOG_DEBUG(
"Hello Data: " << state);
138 for (
const auto& content : state.getContent()) {
139 const ndn::Name& prefix = content.getPrefix(-1);
140 uint64_t seq = content.get(content.size()-1).toNumber();
145 updates.push_back(
MissingDataInfo{prefix, m_prefixes[prefix] + 1, seq});
146 m_prefixes[prefix] = seq;
148 availableSubscriptions.push_back(prefix);
151 m_onReceiveHelloData(availableSubscriptions);
153 if (!updates.empty()) {
154 NDN_LOG_DEBUG(
"Updating application with missed updates");
162 BOOST_ASSERT(!m_iblt.empty());
164 ndn::Name syncInterestName(m_syncInterestPrefix);
170 syncInterestName.append(m_iblt);
172 ndn::Interest syncInterest(syncInterestName);
174 NDN_LOG_DEBUG(
"sendSyncInterest, nonce: " << syncInterest.getNonce() <<
175 " hash: " << std::hash<std::string>{}(syncInterest.getName().toUri()));
178 m_syncFetcher->stop();
181 using ndn::util::SegmentFetcher;
182 SegmentFetcher::Options options;
183 options.interestLifetime = m_syncInterestLifetime;
184 options.maxTimeout = m_syncInterestLifetime;;
185 options.rttOptions.initialRto = m_syncInterestLifetime;
187 m_syncFetcher = SegmentFetcher::start(m_face, syncInterest,
188 ndn::security::v2::getAcceptAllValidator(), options);
190 m_syncFetcher->afterSegmentValidated.connect([
this] (
const ndn::Data& data) {
191 if (data.getFinalBlock()) {
192 m_syncDataName = data.getName().getPrefix(-2);
193 m_syncDataContentType = data.getContentType();
196 if (m_syncDataContentType == ndn::tlv::ContentType_Nack) {
197 NDN_LOG_DEBUG(
"Received application Nack from producer, sending hello again");
202 m_syncFetcher->onComplete.connect([
this] (
const ndn::ConstBufferPtr& bufferPtr) {
203 if (m_syncDataContentType == ndn::tlv::ContentType_Nack) {
204 m_syncDataContentType = ndn::tlv::ContentType_Blob;
207 NDN_LOG_TRACE(
"Segment fetcher got sync data");
208 onSyncData(bufferPtr);
211 m_syncFetcher->onError.connect([
this] (uint32_t errorCode,
const std::string& msg) {
212 NDN_LOG_TRACE(
"Cannot fetch sync data, error: " << errorCode <<
" message: " << msg);
213 ndn::time::milliseconds after(m_rangeUniformRandom(m_rng));
214 NDN_LOG_TRACE(
"Scheduling after " << after);
220 Consumer::onSyncData(
const ndn::ConstBufferPtr& bufferPtr)
223 m_iblt = m_syncDataName.getSubName(m_syncDataName.size() - 1, 1);
225 State state{ndn::Block{bufferPtr}};
227 std::vector<MissingDataInfo> updates;
229 for (
const auto& content : state.getContent()) {
230 NDN_LOG_DEBUG(content);
231 const ndn::Name& prefix = content.getPrefix(-1);
232 uint64_t seq = content.get(content.size() - 1).toNumber();
233 if (m_prefixes.find(prefix) == m_prefixes.end() || seq > m_prefixes[prefix]) {
236 updates.push_back(
MissingDataInfo{prefix, m_prefixes[prefix] + 1, seq});
237 m_prefixes[prefix] = seq;
242 NDN_LOG_DEBUG(
"Sync Data: " << state);
244 if (!updates.empty()) {
Consumer logic to subscribe to producer's data.
NDN_LOG_INIT(psync.Consumer)
Consumer(const ndn::Name &syncPrefix, ndn::Face &face, const ReceiveHelloCallback &onReceiveHelloData, const UpdateCallback &onUpdate, unsigned int count, double false_positive, ndn::time::milliseconds helloInterestLifetime=HELLO_INTEREST_LIFETIME, ndn::time::milliseconds syncInterestLifetime=SYNC_INTEREST_LIFETIME)
constructor
bool addSubscription(const ndn::Name &prefix)
Add prefix to subscription list.
void appendToName(ndn::Name &name) const
Append our bloom filter to the given name.
bool isSubscribed(const ndn::Name &prefix) const
void sendSyncInterest()
send sync interest /<sync-prefix>/sync/<BF>/<producers-IBF>
void stop()
Stop segment fetcher to stop the sync and free resources.
void insert(const std::string &key)
void sendHelloInterest()
send hello interest /<sync-prefix>/hello/
std::function< void(const std::vector< MissingDataInfo > &)> UpdateCallback
std::function< void(const std::vector< ndn::Name > &)> ReceiveHelloCallback