ddtDataSubscriber.hpp

Go to the documentation of this file.

 
#ifndef DDTDATASUBSCRIBER_HPP_
#define DDTDATASUBSCRIBER_HPP_
 
#define BOOST_BIND_GLOBAL_PLACEHOLDERS
 
#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 &log4cplus_logger);

  ~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 SignalT::slot_type &event_listener);
 
 protected:
  void LoadDefaults();

  void StopThreads() override;

  void ReadIni();

  int32_t max_age_data_sample = 0;

  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() override;

  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 = nullptr;
  DdtMemoryAccessor *memory_accessor = nullptr;
  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 = 0;
  std::atomic<bool> event_active{false};
 
  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;

  std::thread subscribe_thread;

  std::thread reregister_thread;
 
  const int32_t NUM_RETRIES = 10;
  const int32_t MAX_AGE_DATA_SAMPLE_MIN = 2000;
};
 
}  // namespace ddt
 
#endif /* DDTDATASUBSCRIBER_HPP_ */