chrono-sync2013.hpp

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

#include <vector>
#include "../face.hpp"
#include "../security/key-chain.hpp"
#include "../util/memory-content-cache.hpp"

namespace google { namespace protobuf { template <typename Element> class RepeatedPtrField; } }
namespace Sync { class SyncStateMsg; }
namespace Sync { class SyncState; }

namespace ndn {

class DigestTree;

class ChronoSync2013 {
public:
  class SyncState;
  typedef func_lib::function<void
    (const std::vector<ChronoSync2013::SyncState>& syncStates, bool isRecovery)>
      OnReceivedSyncState;

  typedef func_lib::function<void()> OnInitialized;

  ChronoSync2013
    (const OnReceivedSyncState& onReceivedSyncState,
     const OnInitialized& onInitialized, const Name& applicationDataPrefix,
     const Name& applicationBroadcastPrefix, int sessionNo,
     Face& face, KeyChain& keyChain, const Name& certificateName,
     Milliseconds syncLifetime, const OnRegisterFailed& onRegisterFailed,
     int previousSequenceNumber = -1)
  : impl_(new Impl
      (onReceivedSyncState, onInitialized, applicationDataPrefix,
       applicationBroadcastPrefix, sessionNo, face, keyChain, certificateName,
       syncLifetime, previousSequenceNumber))
  {
    impl_->initialize(onRegisterFailed);
  }

  class SyncState {
  public:
    SyncState
      (const std::string& dataPrefixUri, int sessionNo, int sequenceNo,
       const Blob& applicationInfo)
    : dataPrefixUri_(dataPrefixUri), sessionNo_(sessionNo),
      sequenceNo_(sequenceNo), applicationInfo_(applicationInfo)
    {
    }

    const std::string&
    getDataPrefix() const { return dataPrefixUri_; }

    int
    getSessionNo() const { return sessionNo_; }

    int
    getSequenceNo() const { return sequenceNo_; }

    const Blob&
    getApplicationInfo() const { return applicationInfo_; }

  private:
    std::string dataPrefixUri_;
    int sessionNo_;
    int sequenceNo_;
    Blob applicationInfo_;
  };

  class PrefixAndSessionNo {
  public:
    PrefixAndSessionNo(const std::string& dataPrefixUri, int sessionNo)
    : dataPrefixUri_(dataPrefixUri), sessionNo_(sessionNo)
    {
    }

    const std::string&
    getDataPrefix() const { return dataPrefixUri_; }

    int
    getSessionNo() const { return sessionNo_; }

  private:
    std::string dataPrefixUri_;
    int sessionNo_;
  };

  void
  getProducerPrefixes(std::vector<PrefixAndSessionNo>& prefixes) const
  {
    impl_->getProducerPrefixes(prefixes);
  }

  int
  getProducerSequenceNo(const std::string& dataPrefix, int sessionNo) const
  {
    return impl_->getProducerSequenceNo(dataPrefix, sessionNo);
  }

  void
  publishNextSequenceNo(const Blob& applicationInfo = Blob())
  {
    return impl_->publishNextSequenceNo(applicationInfo);
  }

  int
  getSequenceNo() const
  {
    return impl_->getSequenceNo();
  }

  void
  shutdown()
  {
    impl_->shutdown();
  }

private:
  class DigestLogEntry {
  public:
    DigestLogEntry
      (const std::string& digest,
       const google::protobuf::RepeatedPtrField<Sync::SyncState>& data);

    const std::string&
    getDigest() const { return digest_; }

    const google::protobuf::RepeatedPtrField<Sync::SyncState>&
    getData() const { return *data_; }

  private:
    std::string digest_;
    ptr_lib::shared_ptr<google::protobuf::RepeatedPtrField<Sync::SyncState> > data_;
  };

  class Impl : public ptr_lib::enable_shared_from_this<Impl> {
  public:
    Impl
      (const OnReceivedSyncState& onReceivedSyncState,
       const OnInitialized& onInitialized, const Name& applicationDataPrefix,
       const Name& applicationBroadcastPrefix, int sessionNo,
       Face& face, KeyChain& keyChain, const Name& certificateName,
       Milliseconds syncLifetime, int previousSequenceNumber);

    void
    initialize(const OnRegisterFailed& onRegisterFailed);

    void
    getProducerPrefixes(std::vector<PrefixAndSessionNo>& prefixes) const;

    int
    getProducerSequenceNo(const std::string& dataPrefix, int sessionNo) const;

    void
    publishNextSequenceNo(const Blob& applicationInfo);

    int
    getSequenceNo() const { return sequenceNo_; }

    void
    shutdown()
    {
      enabled_ = false;
      contentCache_.unregisterAll();
    }

  private:
    void
    broadcastSyncState
      (const std::string& digest, const Sync::SyncStateMsg& syncMessage);

    bool
    update(const google::protobuf::RepeatedPtrField<Sync::SyncState >& content);

    // Search the digest log by digest.
    int
    logFind(const std::string& digest) const;

    void
    onInterest
      (const ptr_lib::shared_ptr<const Name>& prefix,
       const ptr_lib::shared_ptr<const Interest>& interest, Face& face,
       uint64_t registerPrefixId,
       const ptr_lib::shared_ptr<const InterestFilter>& filter);

    // Process Sync Data.
    void
    onData
      (const ptr_lib::shared_ptr<const Interest>& interest,
       const ptr_lib::shared_ptr<Data>& data);

    // Initial sync interest timeout, which means there are no other publishers yet.
    void
    initialTimeOut(const ptr_lib::shared_ptr<const Interest>& interest);

    void
    processRecoveryInterest
      (const Interest& interest, const std::string& syncDigest, Face& face);

    bool
    processSyncInterest(int index, const std::string& syncDigest, Face& face);

    // Send Recovery Interest.
    void
    sendRecovery(const std::string& syncDigest);

    void
    judgeRecovery
      (const ptr_lib::shared_ptr<const Interest> &interest,
       const std::string& syncDigest, Face* face);

    // Sync interest time out, if the interest is the static one send again.
    void
    syncTimeout(const ptr_lib::shared_ptr<const Interest>& interest);

    // Process initial data which usually includes all other publisher's info, and send back the new comer's own info.
    void
    initialOndata(const google::protobuf::RepeatedPtrField<Sync::SyncState >& content);

    static void
    dummyOnData
      (const ptr_lib::shared_ptr<const Interest>& interest,
       const ptr_lib::shared_ptr<Data>& data);

    Face& face_;
    KeyChain& keyChain_;
    Name certificateName_;
    Milliseconds syncLifetime_;
    OnReceivedSyncState onReceivedSyncState_;
    OnInitialized onInitialized_;
    std::vector<ptr_lib::shared_ptr<DigestLogEntry> > digestLog_;
    ptr_lib::shared_ptr<DigestTree> digestTree_;
    std::string applicationDataPrefixUri_;
    const Name applicationBroadcastPrefix_;
    int sessionNo_;
    int initialPreviousSequenceNo_;
    int sequenceNo_;
    MemoryContentCache contentCache_;
    bool enabled_;
  };

  ptr_lib::shared_ptr<Impl> impl_;
};

}

#endif