ddtDataSubscriber.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 ddtDataSubscriber.hpp
// @brief Data Subscriber.
//
// This class provides the functionality for subscriber applications to register
// / unregister at a local broker and to receive data for a specified data
// stream.
//
// @author Matthias Grimm, CGI
// @since  2020/01/16
//
 
#ifndef DDTDATASUBSCRIBER_HPP_
#define DDTDATASUBSCRIBER_HPP_
 
#include <boost/bind.hpp>
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <ddt/ddtStatisticsClient.hpp>
#include "ddt/ddtDataTransferLib.hpp"
 
namespace ddt {
 
class DdtDataSubscriber : public DdtDataTransferLib {
 public:
  explicit DdtDataSubscriber(DdtLogger *logger);
 
  explicit DdtDataSubscriber(log4cplus::Logger const &log4cplusLogger);
 
  ~DdtDataSubscriber() override;
 
  int RegisterSubscriber(const std::string uri, const std::string dsi,
                         const std::string remote_uri,
                         const int32_t interval = 10) override;
 
  int UnregisterSubscriber() override;
 
  DataSample *ReadData() override;
 
  void StartNotificationSubscription();
 
  void StopNotificationSubscription();
 
  DdtStatistics get_statistics();
 
  boost::signals2::connection connect(
      const signal_t::slot_type &event_listener);
 
 protected:
  void LoadDefaults();
 
  void ReadIni();
 
  int32_t max_age_data_sample; 
  const int32_t MAX_AGE_DATA_SAMPLE_DEFAULT = 10000;
 
 private:
  void Init();
 
  void PrintConfigValues();
 
  bool CheckPath();
 
  void InitializeNotificationSubscriber(
      const std::string data_stream_identifier,
      const int32_t notification_port);
 
  void Subscribe();
 
  void Reregister();
 
  void NotificationEvent(
      const mal::ps::DataEvent<datatransfer::NotificationSample> &event);
 
  int CheckPublisher();
 
  int CreateAccessor();
 
  void ReopenShm();
 
  void LogSubscriberParameter();
 
  void PrintErrorMessage(const int result);
 
  DdtStatisticsClient *statistics_client;
  DdtMemoryAccessor *memory_accessor;
  std::string shm_id;
  std::string data_stream_identifier;
  std::string subscriber_uuid;
  std::string broker_uri;
  std::string remote_broker_uri;
  int32_t reading_interval;
  std::atomic<bool> event_active;
 
  std::unique_ptr<mal::ps::Subscriber<datatransfer::NotificationSample>,
                  std::default_delete<
                      mal::ps::Subscriber<datatransfer::NotificationSample> > >
      notification_subscriber;
  std::shared_ptr<datatransfer::NotificationSample> ddt_key_notification;
  std::shared_ptr<datatransfer::NotificationSample> ddt_notification;
  mal::ps::DataEventFilter<datatransfer::NotificationSample> filter;
 
  std::promise<void> exit_signal;
  std::future<void> future_object;
 
  const int32_t NUM_RETRIES = 10;
  const int32_t MAX_AGE_DATA_SAMPLE_MIN = 2000;
};
 
}  // namespace ddt
 
#endif /* DDTDATASUBSCRIBER_HPP_ */