ddtDataTransferLib.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 ddtDataTransferLib.hpp
// @brief Base class for DdtDataPublishers and DdtDataSubscribers.
//
// This is the base class for DdtDataPublishers and DdtDataSubscribers.
//
// @author Matthias Grimm, CGI
// @since  2020/01/16
//
 
#ifndef DDTDATATRANSFERLIB_HPP_
#define DDTDATATRANSFERLIB_HPP_
 
#include <Ddtdatatransfericd.hpp>
#include <boost/property_tree/ini_parser.hpp>
#include <boost/property_tree/ptree.hpp>
#include <iostream>
#include <mal/Cii.hpp>
#include <mal/rr/qos/ReplyTime.hpp>
#include <mal/utility/LoadMal.hpp>
 
#include "ddt/ddtErrorCodes.hpp"
#include "ddt/ddtLogger.hpp"
#include "ddt/ddtMemoryAccessor.hpp"
 
namespace mal = ::elt::mal;
namespace datatransfer = ::elt::ddt::datatransfer;
 
namespace ddt {
 
class DdtDataTransferLib {
 public:
  explicit DdtDataTransferLib(DdtLogger* ddt_logger);
 
  virtual ~DdtDataTransferLib();
 
  void SetQoS(const int ddt_latency, const int ddt_deadline);
 
  virtual int RegisterPublisher(std::string broker_uri,
                                std::string data_stream_identifier,
                                bool compute_checksum) {
    return 0;
  };
 
  virtual int UnregisterPublisher() { return 0; };
 
  virtual void PublishData(){};
 
  virtual int RegisterSubscriber(std::string broker_uri,
                                 std::string data_stream_identifier,
                                 std::string remote_broker_uri,
                                 int32_t reading_interval = 10) {
    return 0;
  };
 
  virtual int UnregisterSubscriber() { return 0; };
 
  virtual DataSample* ReadData() { return nullptr; };
 
 protected:
  void StartHeartbeat(const int32_t interval, const std::string id);
 
  void StopHeartbeat();
 
  void CheckHeartbeatTimeout(int32_t& new_reply_time);
 
  const std::string VerifyPathInBrokerUri(std::string broker_uri);
 
  const std::string GetConfigFilePath();
 
  int latency;  
  int deadline; 
  int32_t reply_time; 
  int32_t heartbeat_interval; 
  std::promise<void> exit_signal_heartbeat;
 
  std::future<void> future_object_heartbeat;
 
  std::atomic<bool> heartbeat_active;
 
  std::unique_ptr<
      datatransfer::DataBrokerRegistrationSync,
      std::default_delete<datatransfer::DataBrokerRegistrationSync> >
      client;
 
  std::atomic<bool> connected_to_broker;
 
  elt::mal::rr::ListenerRegistration connection_listener;
 
  DdtLogger* logger;
 
  const int32_t REPLY_TIME_DEFAULT = 6;
 
  const int32_t REPLY_TIME_MIN = 2;
 
 private:
  void Init(DdtLogger* ddt_logger);
 
  void HeartbeatThread();
 
  std::string identifier;
 
  const std::string BROKER_PATH{"/broker/Broker1"};
  const int LATENCY_DEFAULT = 10000;
  const int DEADLINE_DEFAULT = 10;
  const int32_t HEARTBEAT_INTERVAL_DEFAULT = 1;
};
 
}  // namespace ddt
 
#endif /* DDTDATATRANSFERLIB_HPP_ */