ddtDataConsumer.hpp

Go to the documentation of this file.

// @copyright
// (c) Copyright ESO 2020
// All Rights Reserved
// ESO (eso.org) is an Intergovernmental Organization, and therefore special
// legal conditions apply.
//
// @file ddtDataConsumer.hpp
// @brief Data Consumer.
//
// This class provides the functionality to subscribe to a data stream, to
// write the received data into shared memory and to notify DdtDataSubscribers
// that new data is available. There will be one data consumer object per
// data stream identifier and thus several DdtDataSubscribers may share one
// consumer object.
//
// @author Matthias Grimm, CGI
// @since  2020/01/16
//
 
#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);
 
  virtual ~DdtDataConsumer();
 
  void StartSubscription();
 
  void StopSubscription();
 
  enum SubscriberType {
    LOCAL,
    REMOTE
  };
 
  void AddUuid(std::string uuid, SubscriberType type);
 
  void RemoveUuid(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();
 
  const std::string get_remote_broker_uri();
 
  const int32_t get_notification_port();
 
  DdtStatistics get_statistics();
 
  void ResetStatistics();
 
  const std::string get_publishing_uri();
 
  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* memory_accessor);
 
  void set_notification_port(const int32_t noti_port);
 
  void set_publishing_uri(const std::string pub_uri);
 
  signal_n notification_signal;
 
 private:
  void Init(const std::string& ds_id, DdtLogger* ddt_logger);
 
  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;
 
  DdtMemoryAccessor* memory_accessor;
 
  std::string data_stream_identifier;
  std::string remote_broker_uri;
  int32_t number_of_samples;
  int32_t notification_port;
 
  std::mutex subscriber_mutex;
  std::mutex statistics_mutex;
 
  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; 
  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_ */