ddtClient.hpp

Go to the documentation of this file.

 
#ifndef DDTCLIENT_HPP_
#define DDTCLIENT_HPP_
 
#include <Ddtdatatransfericd.hpp>
#include <atomic>
#include <condition_variable>
#include <mal/Cii.hpp>
#include <mal/rr/qos/ReplyTime.hpp>
#include <mal/utility/LoadMal.hpp>
#include <map>
#include <mutex>
#include <thread>
 
#include "ddt/ddtConstants.hpp"
#include "ddt/ddtLogger.hpp"
 
namespace mal = ::elt::mal;
namespace datatransfer = ::elt::ddt::datatransfer;
 
namespace ddt {

struct DdtClientConfig {
  std::string remote_broker_uri;
  int32_t reply_time;
  int32_t heartbeat_interval;
  int32_t heartbeat_timeout;
  int32_t wait_for_connection;
  int32_t latency;
  int32_t deadline;
  int32_t max_consecutive_failures;
  std::string broker_uri;
};

class DdtClient {
 public:
  explicit DdtClient();

  DdtClient(const DdtClientConfig& config, DdtLogger* ddt_logger);

  virtual ~DdtClient(); 

  void AddUuid(std::string uuid, std::string dsi);

  void UnregisterSubscriber(const std::string uuid);

  bool CheckIfEmpty();

  bool CheckPublisherExists(const std::string& data_stream_identifier) const;

  int32_t RegisterRemoteSubscriber(const std::string& subscriber_uuid,
                                   const std::string& data_stream_identifier,
                                   const int32_t latency,
                                   const int32_t deadline) const;

  std::string GetPublishingUri(const std::string& data_stream_identifier) const;

  int32_t get_max_data_sample_size(
      const std::string& data_stream_identifier) const;

  int32_t get_number_of_samples(
      const std::string& data_stream_identifier) const;

  bool get_compute_checksum(const std::string& data_stream_identifier) const;

  std::string get_broker_uri() const;
 
 protected:
  void Init(const DdtClientConfig& config, DdtLogger* ddt_logger);

  std::map<std::string, std::string> subscriber_map;

  std::atomic<bool> connected_to_broker;

  std::atomic<bool> heartbeat_active;

  std::string remote_broker_uri;

  std::string broker_uri;

  int32_t reply_time;

  int32_t heartbeat_interval;

  int32_t heartbeat_timeout;

  int32_t wait_for_connection;

  DdtLogger* logger;
 
 private:
  std::thread hb_thread;

  std::thread reconnect_thread;

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

  int32_t InitMalClient();

  void StartHeartbeat();

  void StopHeartbeat();

  void HeartbeatThread();

  void Reregister();

  void Reconnect();

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

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

  std::mutex subscriber_mutex;

  std::mutex connection_state_mutex;

  const int32_t NUM_RETRIES = 10;

  int32_t max_consecutive_failures;

  int32_t latency;

  int32_t deadline;
};
 
}  // namespace ddt
 
#endif /* DDTCLIENT_HPP_ */