ddtClient.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 ddtClient.hpp
// @brief Client class for the connection to remote brokers.
//
// This class creates MAL clients to connect to remote brokers. It provides a
// connection listener which observes the connection state and reregisters
// subscribers in the case a broker was restarted. It also provides the
// heartbeat functionality.
//
// @author Matthias Grimm, CGI
// @since  2020/11/18
//
 
#ifndef DDTCLIENT_HPP_
#define DDTCLIENT_HPP_
 
#include <Ddtdatatransfericd.hpp>
#include <future>
#include <mal/Cii.hpp>
#include <mal/rr/qos/ReplyTime.hpp>
#include <mal/utility/LoadMal.hpp>
#include <map>
#include <thread>
 
#include "ddt/ddtConstants.hpp"
#include "ddt/ddtLogger.hpp"
 
namespace mal = ::elt::mal;
namespace datatransfer = ::elt::ddt::datatransfer;
 
namespace ddt {
 
class DdtClient {
 public:
  DdtClient(const std::string remote_broker, const int32_t repl_time,
            const int32_t hb_interval, const int32_t hb_timeout,
            const std::string broker, DdtLogger* ddt_logger);
 
  virtual ~DdtClient();
 
  void AddUuid(std::string uuid, std::string dsi);
 
  void UnregisterSubscriber(std::string uuid);
 
  bool CheckIfEmpty();
 
  bool CheckPublisherExists(const std::string& data_stream_identifier);
 
  int32_t RegisterRemoteSubscriber(const std::string& subscriber_uuid,
                                   const std::string& data_stream_identifier,
                                   const int32_t latency,
                                   const int32_t deadline);
 
  std::string GetPublishingUri(const std::string& data_stream_identifier);
 
  int32_t get_max_data_sample_size(const std::string& data_stream_identifier);
 
  int32_t get_number_of_samples(const std::string& data_stream_identifier);
 
  bool get_compute_checksum(const std::string& data_stream_identifier);
 
 private:
  void Init(const std::string remote_broker, const int32_t repl_time,
            const int32_t hb_interval, const int32_t hb_timeout,
            const std::string broker, DdtLogger* ddt_logger);
 
  int32_t InitMalClient();
 
  void StartHeartbeat();
 
  void StopHeartbeat();
 
  void HeartbeatThread();
 
  void Reregister();
 
  std::unique_ptr<
      datatransfer::DataBrokerRegistrationSync,
      std::default_delete<datatransfer::DataBrokerRegistrationSync> >
      client;
 
  elt::mal::rr::ListenerRegistration connection_listener;
 
  std::atomic<bool> connected_to_broker;
 
  std::map<std::string, std::string>
      subscriber_map; 
  std::promise<void> exit_signal;
  std::future<void> future_object;
  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;
  std::mutex subscriber_mutex;
  DdtLogger* logger;
 
  const int32_t NUM_RETRIES = 10;
  const int32_t LATENCY = 10000;
  const int32_t DEADLINE = 10;
};
 
}  // namespace ddt
 
#endif /* DDTCLIENT_HPP_ */