on_demand_connection_manager.hpp

Go to the documentation of this file.

#ifndef IROHA_ON_DEMAND_CONNECTION_MANAGER_HPP
#define IROHA_ON_DEMAND_CONNECTION_MANAGER_HPP

#include "ordering/on_demand_os_transport.hpp"

#include <atomic>
#include <shared_mutex>

#include <rxcpp/rx-lite.hpp>
#include "logger/logger_fwd.hpp"

namespace iroha {
  namespace ordering {

    class OnDemandConnectionManager : public transport::OdOsNotification {
     public:
      enum PeerType {
        kRejectRejectConsumer = 0,
        kRejectCommitConsumer,
        kCommitRejectConsumer,
        kCommitCommitConsumer,
        kIssuer,
        kCount
      };

      template <typename T>
      using PeerCollectionType = std::array<T, kCount>;

      struct CurrentPeers {
        PeerCollectionType<std::shared_ptr<shared_model::interface::Peer>>
            peers;
      };

      OnDemandConnectionManager(
          std::shared_ptr<transport::OdOsNotificationFactory> factory,
          rxcpp::observable<CurrentPeers> peers,
          logger::LoggerPtr log);

      OnDemandConnectionManager(
          std::shared_ptr<transport::OdOsNotificationFactory> factory,
          rxcpp::observable<CurrentPeers> peers,
          CurrentPeers initial_peers,
          logger::LoggerPtr log);

      ~OnDemandConnectionManager() override;

      void onBatches(CollectionType batches) override;

      boost::optional<std::shared_ptr<const ProposalType>> onRequestProposal(
          consensus::Round round) override;

     private:
      struct CurrentConnections {
        PeerCollectionType<
            boost::optional<std::unique_ptr<transport::OdOsNotification>>>
            peers;
      };

      void initializeConnections(const CurrentPeers &peers);

      logger::LoggerPtr log_;
      std::shared_ptr<transport::OdOsNotificationFactory> factory_;
      rxcpp::composite_subscription subscription_;

      CurrentConnections connections_;

      std::shared_timed_mutex mutex_;
      std::atomic_bool stop_requested_{false};
    };

  }  // namespace ordering
}  // namespace iroha

#endif  // IROHA_ON_DEMAND_CONNECTION_MANAGER_HPP