ddtDataConsumer.hpp

Go to the documentation of this file.

 
#ifndef DDTDATACONSUMER_HPP_
#define DDTDATACONSUMER_HPP_
 
#include "ddt/ddtProducerConsumerBase.hpp"
#include "ddt/ddtStatistics.hpp"
 
namespace mal = ::elt::mal;
namespace datatransfer = ::elt::ddt::datatransfer;

typedef boost::signals2::signal<void(datatransfer::NotificationType,
                                     const std::string&)>
    signal_n;

typedef signal_n::slot_type slot_n;
 
namespace ddt {

class DdtDataConsumer : public DdtProducerConsumerBase {
 public:
  DdtDataConsumer(const std::string& data_stream_identifier,
                  const int32_t latency, const int32_t deadline,
                  DdtLogger* ddt_logger);

  DdtDataConsumer(const std::string& data_stream_identifier,
                  const std::string& subscription_uri, const int32_t latency,
                  const int32_t deadline, DdtLogger* ddt_logger);

  ~DdtDataConsumer() override;

  void StartSubscription();

  void StopSubscription();

  enum SubscriberType {
    LOCAL,
    REMOTE
  };

  void AddUuid(std::string uuid, SubscriberType type);

  void RemoveUuid(const std::string uuid);

  void Notify(const NotificationType type) override;

  int32_t get_number_of_subscribers();

  int32_t get_number_of_remote_subscribers();

  std::map<std::string, SubscriberType> get_subscribers();

  std::string get_remote_broker_uri() const;

  int32_t get_notification_port() const;

  DdtStatistics get_statistics();

  void ResetStatistics();

  std::string get_publishing_uri() const;

  void set_remote_broker_uri(const std::string& remote_uri);

  void set_number_of_samples(const int32_t num_samples);

  void set_memory_accessor(DdtMemoryAccessor* mem_accessor);

  void set_notification_port(const int32_t noti_port);

  void set_publishing_uri(const std::string pub_uri);

  void set_originating_broker(const std::string orig_broker);

  signal_n notification_signal;
 
 protected:
  void Init(const std::string& ds_id, DdtLogger* ddt_logger);

  DdtMemoryAccessor* memory_accessor;

  std::string data_stream_identifier;

  int32_t number_of_samples;

  int32_t notification_port;

  std::string publishing_uri;

  std::chrono::system_clock::time_point last_received;

  uint64_t total_samples = 0;

  uint64_t total_bytes = 0;

  uint64_t total_latency = 0;
 
 private:
  void CreateSubscriber(const std::string& subscription_uri,
                        const int32_t latency, const int32_t deadline);

  void Subscribe();

  void CreateNotifier(const int32_t latency, const int32_t deadline);

  void ReceiveDataEvent(
      const mal::ps::DataEvent<datatransfer::DataPacket>& event);
 
  std::unique_ptr<
      mal::ps::Subscriber<datatransfer::DataPacket>,
      std::default_delete<mal::ps::Subscriber<datatransfer::DataPacket> > >
      data_subscriber;
  mal::ps::DataEventFilter<datatransfer::DataPacket> filter;
  std::shared_ptr<datatransfer::DataPacket> ddt_key_sample;
  std::shared_ptr<datatransfer::DataPacket> ddt_data_packet;
 
  std::string remote_broker_uri;
  std::string originating_broker;
  std::mutex subscriber_mutex;
  std::mutex statistics_mutex;

  std::map<std::string, SubscriberType> subscriber_map;
 
  std::promise<void> exit_signal;
  std::future<void> future_object;

  std::unique_ptr<mal::ps::InstancePublisher<datatransfer::NotificationSample>,
                  std::default_delete<mal::ps::InstancePublisher<
                      datatransfer::NotificationSample> > >
      notifier;
  std::shared_ptr<datatransfer::NotificationSample> ddt_notification_sample;
};
 
}  // namespace ddt
 
#endif /* DDTDATACONSUMER_HPP_ */