notification-subscriber.cpp

Go to the documentation of this file.

/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
#include "notification-subscriber.hpp"
#include "random.hpp"

namespace ndn {
namespace util {

NotificationSubscriberBase::NotificationSubscriberBase(Face& face, const Name& prefix,
                                                       time::milliseconds interestLifetime)
  : m_face(face)
  , m_prefix(prefix)
  , m_isRunning(false)
  , m_lastSequenceNo(std::numeric_limits<uint64_t>::max())
  , m_lastNackSequenceNo(std::numeric_limits<uint64_t>::max())
  , m_attempts(1)
  , m_scheduler(face.getIoService())
  , m_nackEvent(m_scheduler)
  , m_interestLifetime(interestLifetime)
{
}

NotificationSubscriberBase::~NotificationSubscriberBase() = default;

void
NotificationSubscriberBase::start()
{
  if (m_isRunning) // already running
    return;
  m_isRunning = true;

  this->sendInitialInterest();
}

void
NotificationSubscriberBase::stop()
{
  if (!m_isRunning) // not running
    return;
  m_isRunning = false;

  if (m_lastInterestId != 0)
    m_face.removePendingInterest(m_lastInterestId);
  m_lastInterestId = 0;
}

void
NotificationSubscriberBase::sendInitialInterest()
{
  if (this->shouldStop())
    return;

  auto interest = make_shared<Interest>(m_prefix);
  interest->setMustBeFresh(true);
  interest->setChildSelector(1);
  interest->setInterestLifetime(getInterestLifetime());

  m_lastInterestId = m_face.expressInterest(*interest,
                       bind(&NotificationSubscriberBase::afterReceiveData, this, _2),
                       bind(&NotificationSubscriberBase::afterReceiveNack, this, _2),
                       bind(&NotificationSubscriberBase::afterTimeout, this));
}

void
NotificationSubscriberBase::sendNextInterest()
{
  if (this->shouldStop())
    return;

  BOOST_ASSERT(m_lastSequenceNo != std::numeric_limits<uint64_t>::max()); // overflow or missing initial reply

  Name nextName = m_prefix;
  nextName.appendSequenceNumber(m_lastSequenceNo + 1);

  auto interest = make_shared<Interest>(nextName);
  interest->setInterestLifetime(getInterestLifetime());

  m_lastInterestId = m_face.expressInterest(*interest,
                       bind(&NotificationSubscriberBase::afterReceiveData, this, _2),
                       bind(&NotificationSubscriberBase::afterReceiveNack, this, _2),
                       bind(&NotificationSubscriberBase::afterTimeout, this));
}

bool
NotificationSubscriberBase::shouldStop()
{
  if (!m_isRunning)
    return true;
  if (!this->hasSubscriber() && onNack.isEmpty()) {
    this->stop();
    return true;
  }
  return false;
}

void
NotificationSubscriberBase::afterReceiveData(const Data& data)
{
  if (this->shouldStop())
    return;

  try {
    m_lastSequenceNo = data.getName().get(-1).toSequenceNumber();
  }
  catch (const tlv::Error&) {
    this->onDecodeError(data);
    this->sendInitialInterest();
    return;
  }

  if (!this->decodeAndDeliver(data)) {
    this->onDecodeError(data);
    this->sendInitialInterest();
    return;
  }

  this->sendNextInterest();
}

void
NotificationSubscriberBase::afterReceiveNack(const lp::Nack& nack)
{
  if (this->shouldStop())
    return;

  this->onNack(nack);

  time::milliseconds delay = exponentialBackoff(nack);
  m_nackEvent = m_scheduler.scheduleEvent(delay, [this] {this->sendInitialInterest();});
}

void
NotificationSubscriberBase::afterTimeout()
{
  if (this->shouldStop())
    return;

  this->onTimeout();

  this->sendInitialInterest();
}

time::milliseconds
NotificationSubscriberBase::exponentialBackoff(lp::Nack nack)
{
  uint64_t nackSequenceNo;

  try {
    nackSequenceNo = nack.getInterest().getName().get(-1).toSequenceNumber();
  }
  catch (const tlv::Error&) {
    nackSequenceNo = 0;
  }

  if (m_lastNackSequenceNo == nackSequenceNo) {
    ++m_attempts;
  }
  else {
    m_attempts = 1;
  }

  m_lastNackSequenceNo = nackSequenceNo;

  return time::milliseconds(static_cast<uint32_t>(pow(2, m_attempts) * 100 +
                                                  random::generateWord32() % 100));
}

} // namespace util
} // namespace ndn