ddtDataTransferLib.hpp

Go to the documentation of this file.

 
#ifndef DDTDATATRANSFERLIB_HPP_
#define DDTDATATRANSFERLIB_HPP_
 
#include <Ddtdatatransfericd.hpp>
#include <boost/property_tree/ini_parser.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/signals2/signal.hpp>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <mal/Cii.hpp>
#include <mal/rr/qos/ReplyTime.hpp>
#include <mal/utility/LoadMal.hpp>
#include <mutex>
#include <thread>
 
#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:
  enum class ConnectionState {
    Disconnected = 0,
    Connected,
    Reconnecting,
    PublisherDisconnected
  };

  using ConnectionStateSignalT = boost::signals2::signal<void(ConnectionState)>;

  explicit DdtDataTransferLib(DdtLogger* ddt_logger);

  explicit DdtDataTransferLib(log4cplus::Logger const& log4cplus_logger);

  virtual ~DdtDataTransferLib();

  void SetQoS(const int ddt_latency, const int ddt_deadline);

  const std::string VerifyPathInBrokerUri(std::string broker_uri);

  int InitMAL(const std::string broker_uri);

  std::unique_ptr<datatransfer::DataBrokerRegistrationSync,
                  std::default_delete<datatransfer::DataBrokerRegistrationSync>> 
    GetBrokerClient();

  virtual int RegisterPublisher(const std::string uri, const std::string dsi,
                                const bool compute_crc) {
    return 0;
  };

  virtual int UnregisterPublisher() { return 0; };

  virtual void PublishData() {
    // intentionally-blank override
  }

  virtual int RegisterSubscriber(const std::string uri, const std::string dsi,
                                 const std::string remote_uri,
                                 const int32_t interval = 10) {
    return 0;
  };

  virtual int UnregisterSubscriber() { return 0; };

  virtual DataSample* ReadData() { return nullptr; };
 
 
  boost::signals2::connection ConnectToConnectionState(
      const std::function<void(ConnectionState)>& callback);
 
 protected:
  void StartHeartbeat(const int32_t interval, const std::string id);

  void StopHeartbeat();

  virtual void StopThreads();

  void Reconnect();

  virtual void Reregister() {}

  void CheckHeartbeatTimeout(int32_t& new_reply_time);

  void NotifyConnectionState(ConnectionState state);

  const std::string GetConfigFilePath();

  int latency = 0;

  int deadline = 0;

  int32_t reply_time = 0;

  int32_t heartbeat_interval = 0;

  std::thread hb_thread;

  std::thread reconnect_thread;

  std::atomic<bool> reconnect_pending{false};

  void SpawnReconnectionThread();

  std::mutex hb_cv_mutex;
  std::condition_variable hb_cv;

  std::atomic<bool> heartbeat_active{false};

  std::atomic<bool> shutdown_in_progress{false};

  std::unique_ptr<
      datatransfer::DataBrokerRegistrationSync,
      std::default_delete<datatransfer::DataBrokerRegistrationSync> >
      client;

  std::atomic<bool> connected_to_broker{false};

  ConnectionStateSignalT connection_state_signal;

  std::atomic<ConnectionState> last_notified_state{ConnectionState::Disconnected};

  std::string broker_uri;

  elt::mal::rr::ListenerRegistration connection_listener;

  DdtLogger* logger = nullptr;

  DdtLogger* my_logger = nullptr;

  const int32_t REPLY_TIME_DEFAULT = 6;

  const int32_t REPLY_TIME_MIN = 2;

  int32_t wait_for_connection = 0;

  const int32_t WAIT_FOR_CONNECTION_DEFAULT = 10;

  const int32_t WAIT_FOR_CONNECTION_MIN = 2;

  int32_t max_consecutive_failures = 0;

  const int32_t MAX_CONSECUTIVE_FAILURES_DEFAULT = 3;

  const int32_t MAX_CONSECUTIVE_FAILURES_MIN = 1;
 
 private:
  void Init(DdtLogger* ddt_logger);

  const int32_t NUM_RECONNECT_RETRIES = 10;

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