ddtConnectionManager.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 ddtConnectionManager.hpp
// @brief Connection Manager.
//
// This class manages the connection handling between Data Brokers and
// Publisher / Subscriber applications.
//
// @author Matthias Grimm, CGI
// @since  2020/01/16
//
 
#ifndef DDTCONNECTIONMANAGER_HPP_
#define DDTCONNECTIONMANAGER_HPP_
 
#include <boost/bind.hpp>
#include <boost/property_tree/ini_parser.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/signals2/signal.hpp>
#include <boost/bind/bind.hpp>
#include <chrono>
#include <memory>
 
#include <mal/rr/ServerAmi.hpp>
#include <mal/rr/ServerContextProvider.hpp>
#include <mal/rr/qos/ReplyTime.hpp>
 
#include "ddt/ddtLogger.hpp"
 
#include "ddt/ddtClient.hpp"
#include "ddt/ddtDataConsumer.hpp"
#include "ddt/ddtDataProducer.hpp"
#include "ddt/ddtMemoryManager.hpp"
 
namespace mal = ::elt::mal;
namespace datatransfer = ::elt::ddt::datatransfer;
 
namespace ddt {
 
class DdtConnectionManager
    : public virtual datatransfer::DataBrokerRegistration {
 public:
  explicit DdtConnectionManager(DdtLogger* ddt_logger,
                                const std::string uri_string,
                                const std::string config);
 
  explicit DdtConnectionManager(DdtMemoryManager* const mmgr,
                                DdtLogger* ddt_logger,
                                const std::string uri_string,
                                const std::string config);
 
  ~DdtConnectionManager() override;
 
  int32_t RegisterPublisher(const std::string& data_stream_identifier,
                            const int32_t latency, int32_t deadline,
                            const int32_t max_data_sample_size,
                            const int32_t number_of_samples,
                            const bool compute_checksum,
                            const std::string& publishing_uri) override;
 
  int32_t UnregisterPublisher(
      const std::string& data_stream_identifier) override;
 
  int32_t UnregisterPublishers();
 
  void PublishData(const std::string& data_stream_identifier) override;
 
  int32_t RegisterSubscriber(const std::string& subscriber_uuid,
                             const std::string& data_stream_identifier,
                             const std::string& remote_broker_uri,
                             const int32_t latency,
                             const int32_t deadline) override;
 
  int32_t RegisterRemoteSubscriber(const std::string& remote_broker,
                                   const std::string& subscriber_uuid,
                                   const std::string& data_stream_identifier,
                                   const int32_t latency,
                                   const int32_t deadline) override;
 
  int32_t UnregisterSubscriber(const std::string& data_stream_identifier,
                               const std::string& subscriber_uuid) override;
 
  int32_t UnregisterSubscribers();
 
  int32_t get_max_data_sample_size(
      const std::string& data_stream_identifier) override;
 
  int32_t get_number_of_samples(
      const std::string& data_stream_identifier) override;
 
  std::string get_publishing_uri(
      const std::string& data_stream_identifier) override;
 
  int32_t get_notification_port(
      const std::string& data_stream_identifier) override;
 
  std::vector<std::string> get_statistics(
      const std::string& data_stream_identifier) override;
 
  int32_t get_heartbeat_interval() override;
 
  int32_t get_heartbeat_timeout() override;
 
  bool get_compute_checksum(const std::string& data_stream_identifier) override;
 
  std::string get_shm_id(const std::string& data_stream_identifier) override;
 
  void UpdateHeartbeat(const std::string& identifier) override;
 
  bool CheckPubRegistrationPermitted(
      const std::string& data_stream_identifier) override;
 
  bool CheckPublisherExists(const std::string& data_stream_identifier) override;
 
  bool CheckRemotePublisherExists(
      const std::string& remote_broker_uri,
      const std::string& data_stream_identifier) override;
 
  void UpdateStatistics(const std::string& data_stream_identifier,
                        const int32_t datavec_size,
                        const uint64_t source_timestamp) override;
 
  int32_t GetMaxPossibleBufferSize(int32_t max_data_sample_size) override;
 
  std::vector<std::string> GetRegisteredStreams() override;
 
  std::vector<std::string> GetConnectedBrokers() override;
 
 protected:
  void LoadDefaults();
 
  std::string GetConfigPath() const;
 
  void ReadIni();
 
  std::string CreateSubscriptionUri(const std::string publishing_uri,
                                    const std::string remote_broker_uri) const;
 
  bool CheckStreamIdInUse(const std::string& data_stream_identifier);
 
  void CreateStatistics(const std::string& data_stream_identifier,
                        const int32_t number_of_samples);
 
  void ResetStatistics(const std::string& data_stream_identifier);
 
  const int32_t SHM_TIMEOUT_DEFAULT = 10;
  const int32_t WAITING_TIME_DEFAULT = 1000;
  const int32_t REPLY_TIME_DEFAULT = 6;
  const int32_t HEARTBEAT_INTERVAL_DEFAULT = 1;
  const int32_t HEARTBEAT_TIMEOUT_DEFAULT = 10;
 
  int32_t shm_timeout;   // configurable timeout after which a shared memory is
                         // deleted in [s]
  int32_t waiting_time;  // configurable waiting time for MAL publishers to
                         // establish communication in [ms]
  int32_t reply_time;    // configurable reply time for MAL clients in [s]
  int32_t heartbeat_interval;  // configurable interval for the heartbeat in [s]
  int32_t heartbeat_timeout;   // configurable timeout for the heartbeat in [s]
 
  std::set<std::string> registered_publishers;
  std::map<std::string, DdtStatistics>
      statistics_map; 
 private:
  void Init(DdtMemoryManager* mmgr, DdtLogger* ddt_logger,
            const std::string uri_string, const std::string config);
 
  void PrintConfigValues();
 
  void StartHeartbeat();
 
  void StopHeartbeat();
 
  void HeartbeatThread();
 
  void NotifySubscribers(const std::string& data_stream_identifier,
                         std::unique_lock<std::mutex>& producer_lock);
 
  int32_t CheckPublisherUsingStream(const std::string data_stream_identifier,
                                    const std::string remote_broker_uri);
 
  void CreateDataConsumer(std::unique_lock<std::mutex>& consumer_lock,
                          const std::string subscriber_uuid,
                          const std::string data_stream_identifier,
                          const std::string remote_broker_uri,
                          const int32_t latency, const int32_t deadline,
                          const std::string subscription_uri,
                          const int32_t number_of_samples);
 
  void SearchAndUnregSubscriber(const std::string identifier);
 
  int32_t RegisterLocalSubscriber(const std::string subscriber_uuid,
                                  const std::string data_stream_identifier,
                                  const int32_t latency,
                                  const int32_t deadline);
 
  int32_t RegisterLocalSubscriberRemote(
      const std::string subscriber_uuid,
      const std::string data_stream_identifier,
      const std::string remote_broker_uri, const int32_t latency,
      const int32_t deadline);
 
  void ProcessNotificationEvent(const datatransfer::NotificationType type,
                                const std::string& data_stream_identifier);
 
  void FreeShmThread(const std::string& data_stream_identifier,
                     const int shm_timeout);
 
  bool CheckPublisherReregistration(const std::string& data_stream_identifier);
 
  bool CheckSharedMemoryRecreation(const std::string& data_stream_identifier,
                                   const int32_t max_data_sample_size,
                                   const int32_t number_of_samples);
 
  void Publish(const std::string& data_stream_identifier);
 
  void PubRegNotification(const std::string& data_stream_identifier);
 
  void PubUnregNotification(const std::string& data_stream_identifier);
 
  std::map<std::string, DdtDataProducer*>
      producer_map; 
  std::map<std::string, DdtDataConsumer*>
      consumer_map; 
  std::map<std::string,
           std::chrono::time_point<std::chrono::high_resolution_clock>>
      client_map;
  std::map<std::string, std::string>
      connected_brokers_map; 
  DdtMemoryManager* memory_manager;
 
  std::map<std::string, DdtClient*>
      ddt_clients; 
  std::atomic<bool> stop_threads;
  std::atomic<int> thread_counter;
 
  std::mutex producer_mutex;
  std::mutex consumer_mutex;
  std::mutex client_mutex;
  std::mutex statistics_mutex;
  std::mutex ddt_clients_mutex;
  std::mutex registered_publishers_mutex;
  std::mutex connected_brokers_mutex;
 
  DdtLogger* logger;
 
  std::promise<void> exit_signal;
  std::future<void> future_object;
  std::atomic<bool> heartbeat_active;
  std::string broker_uri;
  std::string config_file;
 
  boost::signals2::connection connection;
 
  const int32_t SHM_TIMEOUT_MIN = 2;
  const int32_t WAITING_TIME_MIN = 1000;
  const int32_t REPLY_TIME_MIN = 2;
  const int32_t HEARTBEAT_INTERVAL_MIN = 0;
  const int32_t HEARTBEAT_TIMEOUT_MIN = 3;
  const int32_t NUM_RETRIES = 30;
};
 
}  // namespace ddt
 
#endif /* DDTCONNECTIONMANAGER_HPP_ */