connection_manager.hpp

Go to the documentation of this file.

// Copyright (c) 2021 MIT Digital Currency Initiative,
//                    Federal Reserve Bank of Boston
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
#ifndef OPENCBDC_TX_SRC_NETWORK_CONNECTION_MANAGER_H_
#define OPENCBDC_TX_SRC_NETWORK_CONNECTION_MANAGER_H_
 
#include "peer.hpp"
#include "socket_selector.hpp"
#include "tcp_listener.hpp"
#include "tcp_socket.hpp"
#include "util/common/config.hpp"
 
#include <atomic>
#include <cassert>
#include <condition_variable>
#include <mutex>
#include <queue>
#include <random>
#include <shared_mutex>
#include <sys/socket.h>
#include <thread>
 
namespace cbdc::network {
    using peer_id_t = size_t;
 
    struct message_t {
        std::shared_ptr<buffer> m_pkt;
        peer_id_t m_peer_id{};
    };
 
    using packet_handler_t = std::function<std::optional<buffer>(message_t&&)>;
 
    class connection_manager {
      public:
        connection_manager() = default;
 
        connection_manager(const connection_manager&) = delete;
        auto operator=(const connection_manager&)
            -> connection_manager& = delete;
 
        connection_manager(connection_manager&&) = delete;
        auto operator=(connection_manager&&) -> connection_manager& = delete;
 
        ~connection_manager();
 
        [[nodiscard]] auto listen(const ip_address& host, unsigned short port)
            -> bool;
 
        [[nodiscard]] auto pump() -> bool;
 
        void broadcast(const std::shared_ptr<buffer>& data);
 
        template<typename Ta>
        void broadcast(const Ta& data) {
            auto pkt = make_shared_buffer(data);
            return broadcast(pkt);
        }
 
        [[nodiscard]] auto handle_messages() -> std::vector<message_t>;
 
        auto add(std::unique_ptr<tcp_socket> sock,
                 bool attempt_reconnect = true) -> peer_id_t;
 
        auto cluster_connect(const std::vector<endpoint_t>& endpoints,
                             bool error_fatal = true) -> bool;
 
        [[nodiscard]] auto
        start_cluster_handler(const std::vector<endpoint_t>& endpoints,
                              const packet_handler_t& handler)
            -> std::optional<std::thread>;
 
        [[nodiscard]] auto start_server(const endpoint_t& listen_endpoint,
                                        const packet_handler_t& handler)
            -> std::optional<std::thread>;
 
        [[nodiscard]] auto start_server_listener() -> std::thread;
 
        [[nodiscard]] auto start_handler(const packet_handler_t& handler)
            -> std::thread;
 
        void close();
 
        void send(const std::shared_ptr<buffer>& data, peer_id_t peer_id);
 
        template<typename Ta>
        void send(const Ta& data, peer_id_t peer_id) {
            auto pkt = make_shared_buffer(data);
            return send(pkt, peer_id);
        }
 
        [[nodiscard]] auto peer_count() -> size_t;
 
        void reset();
 
        [[nodiscard]] auto send_to_one(const std::shared_ptr<buffer>& data)
            -> bool;
 
        template<typename T>
        [[nodiscard]] auto send_to_one(const T& data) -> bool {
            auto pkt = make_shared_buffer(data);
            return send_to_one(pkt);
        }
 
        [[nodiscard]] auto connected(peer_id_t peer_id) -> bool;
 
        [[nodiscard]] auto connected_to_one() -> bool;
 
      private:
        tcp_listener m_listener;
 
        struct m_peer_t {
            m_peer_t() = delete;
 
            m_peer_t(std::unique_ptr<peer> peer, peer_id_t peer_id);
            ~m_peer_t() = default;
 
            auto operator=(const m_peer_t& other) -> m_peer_t& = default;
            m_peer_t(const m_peer_t& other) = default;
 
            auto operator=(m_peer_t&& other) noexcept -> m_peer_t& = default;
            m_peer_t(m_peer_t&& other) noexcept = default;
 
            std::shared_ptr<peer> m_peer;
            peer_id_t m_peer_id;
        };
 
        std::vector<m_peer_t> m_peers;
        std::atomic<peer_id_t> m_next_peer_id{0};
 
        std::shared_mutex m_peer_mutex;
 
        std::atomic_bool m_running{true};
 
        std::mutex m_async_recv_mut;
        std::condition_variable m_async_recv_cv;
        std::vector<std::queue<message_t>> m_async_recv_queues;
        bool m_async_recv_data{false};
 
        socket_selector m_listen_selector;
 
        std::random_device m_r{cbdc::config::random_source};
        std::default_random_engine m_rnd{m_r()};
    };
}
 
#endif // OPENCBDC_TX_SRC_NETWORK_CONNECTION_MANAGER_H_