37 time::milliseconds interestLifetime)
41 , m_lastSequenceNo(
std::numeric_limits<uint64_t>::max())
42 , m_lastNackSequenceNo(
std::numeric_limits<uint64_t>::max())
44 , m_scheduler(face.getIoService())
45 , m_nackEvent(m_scheduler)
46 , m_lastInterestId(nullptr)
47 , m_interestLifetime(interestLifetime)
60 sendInitialInterest();
70 if (m_lastInterestId !=
nullptr)
72 m_lastInterestId =
nullptr;
76 NotificationSubscriberBase::sendInitialInterest()
81 auto interest = make_shared<Interest>(m_prefix);
82 interest->setCanBePrefix(
true);
83 interest->setMustBeFresh(
true);
84 interest->setChildSelector(1);
88 [
this] (
const auto&,
const auto& d) { this->afterReceiveData(d); },
89 [
this] (
const auto&,
const auto& n) { this->afterReceiveNack(n); },
90 [
this] (
const auto&) { this->afterTimeout(); });
94 NotificationSubscriberBase::sendNextInterest()
99 BOOST_ASSERT(m_lastSequenceNo != std::numeric_limits<uint64_t>::max());
101 Name nextName = m_prefix;
104 auto interest = make_shared<Interest>(nextName);
105 interest->setCanBePrefix(
false);
109 [
this] (
const auto&,
const auto& d) { this->afterReceiveData(d); },
110 [
this] (
const auto&,
const auto& n) { this->afterReceiveNack(n); },
111 [
this] (
const auto&) { this->afterTimeout(); });
115 NotificationSubscriberBase::shouldStop()
120 if (!hasSubscriber() &&
onNack.isEmpty()) {
128 NotificationSubscriberBase::afterReceiveData(
const Data& data)
138 sendInitialInterest();
142 if (!decodeAndDeliver(data)) {
144 sendInitialInterest();
152 NotificationSubscriberBase::afterReceiveNack(
const lp::Nack& nack)
159 time::milliseconds delay = exponentialBackoff(nack);
160 m_nackEvent = m_scheduler.
scheduleEvent(delay, [
this] { sendInitialInterest(); });
164 NotificationSubscriberBase::afterTimeout()
171 sendInitialInterest();
175 NotificationSubscriberBase::exponentialBackoff(
lp::Nack nack)
177 uint64_t nackSequenceNo;
185 if (m_lastNackSequenceNo == nackSequenceNo) {
192 m_lastNackSequenceNo = nackSequenceNo;
194 return time::milliseconds(static_cast<time::milliseconds::rep>(std::pow(2, m_attempts) * 100 +
void start()
start or resume receiving notifications
const Name & getName() const
Copyright (c) 2013-2017 Regents of the University of California.
virtual ~NotificationSubscriberBase()
time::milliseconds getInterestLifetime() const
const Interest & getInterest() const
EventId scheduleEvent(time::nanoseconds after, const EventCallback &callback)
Schedule a one-time event after the specified delay.
represents a Network Nack
uint32_t generateWord32()
Generate a non-cryptographically-secure random integer in the range [0, 2^32)
signal::Signal< NotificationSubscriberBase, lp::Nack > onNack
fires when a NACK is received
signal::Signal< NotificationSubscriberBase > onTimeout
fires when no Notification is received within .getInterestLifetime period
Provide a communication channel with local or remote NDN forwarder.
Name & appendSequenceNumber(uint64_t seqNo)
Append a sequence number component.
Represents an absolute name.
const Name & getName() const
Get name.
signal::Signal< NotificationSubscriberBase, Data > onDecodeError
fires when a Data packet in the Notification Stream cannot be decoded as Notification ...
uint64_t toSequenceNumber() const
Interpret as sequence number component using NDN naming conventions.
const PendingInterestId * expressInterest(const Interest &interest, const DataCallback &afterSatisfied, const NackCallback &afterNacked, const TimeoutCallback &afterTimeout)
Express Interest.
Represents a Data packet.
const Component & get(ssize_t i) const
Get the component at the given index.
void removePendingInterest(const PendingInterestId *pendingInterestId)
Cancel previously expressed Interest.
represents an error in TLV encoding or decoding
NotificationSubscriberBase(Face &face, const Name &prefix, time::milliseconds interestLifetime)
construct a NotificationSubscriber
void stop()
stop receiving notifications