notification-subscriber.hpp

Go to the documentation of this file.

/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
#ifndef NDN_UTIL_NOTIFICATION_SUBSCRIBER_HPP
#define NDN_UTIL_NOTIFICATION_SUBSCRIBER_HPP

#include "../face.hpp"
#include "signal.hpp"
#include "concepts.hpp"
#include "time.hpp"
#include "random.hpp"
#include "scheduler.hpp"
#include "scheduler-scoped-event-id.hpp"
#include <boost/concept_check.hpp>

namespace ndn {
namespace util {

template<typename Notification>
class NotificationSubscriber : noncopyable
{
public:
  BOOST_CONCEPT_ASSERT((boost::DefaultConstructible<Notification>));
  BOOST_CONCEPT_ASSERT((WireDecodable<Notification>));

  NotificationSubscriber(Face& face, const Name& prefix,
                         const time::milliseconds& interestLifetime = time::milliseconds(60000))
    : 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)
  {
  }

  virtual
  ~NotificationSubscriber()
  {
  }

  time::milliseconds
  getInterestLifetime() const
  {
    return m_interestLifetime;
  }

  bool
  isRunning() const
  {
    return m_isRunning;
  }

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

    this->sendInitialInterest();
  }

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

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

public: // subscriptions
  signal::Signal<NotificationSubscriber, Notification> onNotification;

  signal::Signal<NotificationSubscriber, lp::Nack> onNack;

  signal::Signal<NotificationSubscriber> onTimeout;

  signal::Signal<NotificationSubscriber, Data> onDecodeError;

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

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

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

  void
  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);

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

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

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

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

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

    this->onNotification(notification);

    this->sendNextInterest();
  }

  void
  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
  afterTimeout()
  {
    if (this->shouldStop())
      return;

    this->onTimeout();

    this->sendInitialInterest();
  }

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

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

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

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

    m_lastNackSequenceNo = nackSequenceNo;
    return delayTime;
  }

private:
  Face& m_face;
  Name m_prefix;
  bool m_isRunning;
  uint64_t m_lastSequenceNo;
  uint64_t m_lastNackSequenceNo;
  uint64_t m_attempts;
  util::scheduler::Scheduler m_scheduler;
  util::scheduler::ScopedEventId m_nackEvent;
  const PendingInterestId* m_lastInterestId;
  time::milliseconds m_interestLifetime;
};

} // namespace util
} // namespace ndn

#endif // NDN_UTIL_NOTIFICATION_SUBSCRIBER_HPP