segment-consumer.hpp

Go to the documentation of this file.

#ifndef NDNPH_APP_SEGMENT_CONSUMER_HPP
#define NDNPH_APP_SEGMENT_CONSUMER_HPP
 
#include "../face/packet-handler.hpp"
#include "../keychain/null.hpp"
#include "../port/clock/port.hpp"
 
namespace ndnph {
 
class SegmentConsumerBase : public PacketHandler {
public:
  struct Options {
    const PublicKey& verifier = NullKey::get();
 
    int retxLimit = 5;
 
    int retxDelay = 500;
  };
 
  explicit SegmentConsumerBase(Face& face, Options opts)
    : PacketHandler(face)
    , m_opts(std::move(opts))
    , m_pending(this) {}
 
  explicit SegmentConsumerBase(Face& face)
    : SegmentConsumerBase(face, Options()) {}
 
  using SegmentCallback = void (*)(void* ctx, uint64_t segment, Data data);
 
  void setSegmentCallback(SegmentCallback cb, void* ctx) {
    m_cb = cb;
    m_cbCtx = ctx;
  }
 
  class SaveDest {
  public:
    explicit SaveDest(uint8_t* output, size_t limit)
      : output(output)
      , limit(limit) {}
 
    static void accumulate(void* self0, uint64_t, Data data) {
      reinterpret_cast<SaveDest*>(self0)->accumulate(data);
    }
 
  private:
    void accumulate(Data data) {
      if (hasError || !data) {
        hasError = true;
        return;
      }
      auto content = data.getContent();
      if (length + content.size() > limit) {
        hasError = true;
        return;
      }
      std::copy(content.begin(), content.end(), &output[length]);
      length += content.size();
      isCompleted = data.getIsFinalBlock();
    }
 
  public:
    uint8_t* output = nullptr;
    size_t limit = 0;
    size_t length = 0;
    bool isCompleted = false;
    bool hasError = false;
  };
 
  void saveTo(SaveDest& dest) {
    setSegmentCallback(SaveDest::accumulate, &dest);
  }
 
  void start(Name prefix) {
    m_prefix = prefix;
    m_running = true;
    m_segment = 0;
    m_pending.expireNow();
    m_retxRemain = m_opts.retxLimit;
  }
 
  void stop() {
    m_running = false;
  }
 
  bool isRunning() const {
    return m_running;
  }
 
protected:
  void invokeCallback(Data data) {
    if (m_cb != nullptr) {
      m_cb(m_cbCtx, m_segment, data);
    }
  }
 
protected:
  Options m_opts;
  SegmentCallback m_cb = nullptr;
  void* m_cbCtx = nullptr;
  Name m_prefix;
  uint64_t m_segment = 0;
  OutgoingPendingInterest m_pending;
  int m_retxRemain = 0;
  bool m_running = false;
};
 
template<typename SegmentConvention = convention::Segment, size_t regionCap = 1024>
class BasicSegmentConsumer : public SegmentConsumerBase {
public:
  using SegmentConsumerBase::SegmentConsumerBase;
 
private:
  void loop() final {
    if (!m_running || !m_pending.expired()) {
      return;
    }
 
    if (--m_retxRemain < -1) {
      m_running = false;
      invokeCallback(Data());
      return;
    }
 
    StaticRegion<regionCap> region;
    Interest interest = region.template create<Interest>();
    NDNPH_ASSERT(!!interest);
    interest.setName(m_prefix.append(region, SegmentConvention(), m_segment));
    m_pending.send(interest, m_opts.retxDelay);
  }
 
  bool processData(Data data) final {
    StaticRegion<regionCap> region;
    Name interestName = m_prefix.append(region, SegmentConvention(), m_segment);
    if (!m_pending.match(data, interestName, false) || !data.verify(m_opts.verifier)) {
      return false;
    }
 
    invokeCallback(data);
 
    if (data.getIsFinalBlock()) {
      m_running = false;
    } else {
      ++m_segment;
      m_pending.expireNow();
      m_retxRemain = m_opts.retxLimit;
    }
    return true;
  }
};
 
using SegmentConsumer = BasicSegmentConsumer<>;
 
} // namespace ndnph
 
#endif // NDNPH_APP_SEGMENT_CONSUMER_HPP