ddtConnectionManager.hpp

Go to the documentation of this file.

 
#ifndef DDTCONNECTIONMANAGER_HPP_
#define DDTCONNECTIONMANAGER_HPP_
 
#define BOOST_BIND_GLOBAL_PLACEHOLDERS
 
#include <boost/bind/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 <mal/rr/ServerAmi.hpp>
#include <mal/rr/ServerContextProvider.hpp>
#include <mal/rr/qos/ReplyTime.hpp>
#include <memory>
 
#include "ddt/ddtClient.hpp"
#include "ddt/ddtDataConsumer.hpp"
#include "ddt/ddtDataProducer.hpp"
#include "ddt/ddtLogger.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;

  std::string get_shm_full_path(const std::string& data_stream_identifier) override;

  std::string get_broker_uri() 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;

  const int32_t WAIT_FOR_CONNECTION_DEFAULT = 10;

  const int32_t LATENCY_DEFAULT = 10000;

  const int32_t DEADLINE_DEFAULT = 10;

  const int32_t MAX_CONSECUTIVE_FAILURES_DEFAULT = 3;

  int32_t shm_timeout;

  int32_t waiting_time;

  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::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 std::string originating_broker,
                          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 WAIT_FOR_CONNECTION_MIN = 2;

  const int32_t LATENCY_MIN = 1000;

  const int32_t DEADLINE_MIN = 1;

  const int32_t MAX_CONSECUTIVE_FAILURES_MIN = 1;

  const int32_t NUM_RETRIES = 30;
};
 
}  // namespace ddt
 
#endif /* DDTCONNECTIONMANAGER_HPP_ */