41 , m_lastSequenceNum(std::numeric_limits<uint64_t>::max())
42 , m_lastNackSequenceNum(std::numeric_limits<uint64_t>::max())
44 , m_scheduler(face.getIoService())
45 , m_interestLifetime(interestLifetime)
58 sendInitialInterest();
72 NotificationSubscriberBase::sendInitialInterest()
77 auto interest = make_shared<Interest>(m_prefix);
78 interest->setCanBePrefix(
true);
79 interest->setMustBeFresh(
true);
80 interest->setInterestLifetime(m_interestLifetime);
81 sendInterest(*interest);
85 NotificationSubscriberBase::sendNextInterest()
90 Name nextName = m_prefix;
93 auto interest = make_shared<Interest>(nextName);
94 interest->setInterestLifetime(m_interestLifetime);
95 sendInterest(*interest);
99 NotificationSubscriberBase::sendInterest(
const Interest& interest)
102 [
this] (
const auto&,
const auto& d) { this->afterReceiveData(d); },
103 [
this] (
const auto&,
const auto& n) { this->afterReceiveNack(n); },
104 [
this] (
const auto&) { this->afterTimeout(); });
108 NotificationSubscriberBase::shouldStop()
113 if (!hasSubscriber() &&
onNack.isEmpty()) {
121 NotificationSubscriberBase::afterReceiveData(
const Data& data)
127 m_lastSequenceNum = data.getName().get(-1).toSequenceNumber();
129 catch (
const tlv::Error&) {
131 sendInitialInterest();
135 if (!decodeAndDeliver(data)) {
137 sendInitialInterest();
145 NotificationSubscriberBase::afterReceiveNack(
const lp::Nack& nack)
153 m_nackEvent = m_scheduler.
schedule(delay, [
this] { sendInitialInterest(); });
157 NotificationSubscriberBase::afterTimeout()
164 sendInitialInterest();
168 NotificationSubscriberBase::exponentialBackoff(
lp::Nack nack)
170 uint64_t nackSequenceNum;
172 nackSequenceNum = nack.getInterest().getName().get(-1).toSequenceNumber();
174 catch (
const tlv::Error&) {
178 if (m_lastNackSequenceNum == nackSequenceNum) {
185 m_lastNackSequenceNum = nackSequenceNum;
187 return time::milliseconds(
static_cast<time::milliseconds::rep
>(std::pow(2, m_attempts) * 100 +
Provide a communication channel with local or remote NDN forwarder.
PendingInterestHandle expressInterest(const Interest &interest, const DataCallback &afterSatisfied, const NackCallback &afterNacked, const TimeoutCallback &afterTimeout)
Express an Interest.
Represents an absolute name.
Name & appendSequenceNumber(uint64_t seqNo)
Append a sequence number component.
void cancel()
Cancel the operation.
EventId schedule(time::nanoseconds after, EventCallback callback)
Schedule a one-time event after the specified delay.
void start()
Start or resume receiving notifications.
signal::Signal< NotificationSubscriberBase, Data > onDecodeError
Fires when a Data packet in the Notification Stream cannot be decoded as Notification.
void stop()
Stop receiving notifications.
virtual ~NotificationSubscriberBase()
signal::Signal< NotificationSubscriberBase > onTimeout
Fires when no Notification is received within getInterestLifetime() period.
NotificationSubscriberBase(Face &face, const Name &prefix, time::milliseconds interestLifetime)
Construct a NotificationSubscriber.
signal::Signal< NotificationSubscriberBase, lp::Nack > onNack
Fires when a Nack is received.
uint32_t generateWord32()
Generate a non-cryptographically-secure random integer in the range [0, 2^32)
boost::chrono::milliseconds milliseconds