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_ */