messaging_server.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2024, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
#ifndef KCENON_NETWORK_INTERNAL_CORE_MESSAGING_SERVER_H_
#define KCENON_NETWORK_INTERNAL_CORE_MESSAGING_SERVER_H_
 
#include <kcenon/network/detail/config/feature_flags.h>
 
#include <atomic>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <string>
#include <vector>
 
#include <asio.hpp>
 
#include "internal/core/callback_indices.h"
#include "kcenon/network/detail/utils/result_types.h"
#include "kcenon/network/detail/utils/startable_base.h"
#include "kcenon/network/detail/utils/callback_manager.h"
#include "internal/integration/io_context_thread_manager.h"
 
// Optional monitoring support via common_system
#if KCENON_WITH_COMMON_SYSTEM
    #include <kcenon/common/interfaces/monitoring_interface.h>
#endif // KCENON_WITH_COMMON_SYSTEM
 
namespace kcenon::network::session {
    class messaging_session;
}
 
namespace kcenon::network::core {
 
    class messaging_server
        : public std::enable_shared_from_this<messaging_server>
        , public utils::startable_base<messaging_server>
    {
        friend class utils::startable_base<messaging_server>;
    public:
        using connection_callback_t = std::function<void(std::shared_ptr<session::messaging_session>)>;
        using disconnection_callback_t = std::function<void(const std::string&)>;
        using receive_callback_t = std::function<void(std::shared_ptr<session::messaging_session>,
                                                      const std::vector<uint8_t>&)>;
        using error_callback_t = std::function<void(std::shared_ptr<session::messaging_session>,
                                                    std::error_code)>;
 
        explicit messaging_server(const std::string& server_id);
 
        ~messaging_server() noexcept;
 
        // Non-copyable, non-movable
        messaging_server(const messaging_server&) = delete;
        messaging_server& operator=(const messaging_server&) = delete;
        messaging_server(messaging_server&&) = delete;
        messaging_server& operator=(messaging_server&&) = delete;
 
        // =====================================================================
        // Lifecycle Management
        // =====================================================================
 
        [[nodiscard]] auto start_server(unsigned short port) -> VoidResult;
 
        [[nodiscard]] auto stop_server() -> VoidResult;
 
        // Note: wait_for_stop() and is_running() are inherited from startable_base
 
        [[nodiscard]] auto server_id() const -> const std::string&;
 
        // =====================================================================
        // Callback Setters
        // =====================================================================
 
        auto set_connection_callback(connection_callback_t callback) -> void;
 
        auto set_disconnection_callback(disconnection_callback_t callback) -> void;
 
        auto set_receive_callback(receive_callback_t callback) -> void;
 
        auto set_error_callback(error_callback_t callback) -> void;
 
#if KCENON_WITH_COMMON_SYSTEM
        auto set_monitor(kcenon::common::interfaces::IMonitor* monitor) -> void;
 
        auto get_monitor() const -> kcenon::common::interfaces::IMonitor*;
#endif // KCENON_WITH_COMMON_SYSTEM
 
    private:
        // =====================================================================
        // startable_base CRTP interface
        // =====================================================================
 
        [[nodiscard]] static constexpr auto component_name() noexcept -> std::string_view
        {
            return "Server";
        }
 
        auto on_stopped() -> void;
 
        // =====================================================================
        // Internal Implementation Methods
        // =====================================================================
 
        auto do_start_impl(unsigned short port) -> VoidResult;
 
        auto do_stop_impl() -> VoidResult;
 
        // =====================================================================
        // Internal Callback Helpers
        // =====================================================================
 
        [[nodiscard]] auto get_connection_callback() const -> connection_callback_t;
 
        [[nodiscard]] auto get_disconnection_callback() const -> disconnection_callback_t;
 
        [[nodiscard]] auto get_receive_callback() const -> receive_callback_t;
 
        [[nodiscard]] auto get_error_callback() const -> error_callback_t;
 
        auto invoke_connection_callback(std::shared_ptr<session::messaging_session> session) -> void;
 
        // =====================================================================
        // Internal Connection Handlers
        // =====================================================================
 
        auto do_accept() -> void;
 
        auto on_accept(std::error_code ec, asio::ip::tcp::socket socket)
            -> void;
 
        auto cleanup_dead_sessions() -> void;
 
        auto start_cleanup_timer() -> void;
 
    private:
        using callback_index = tcp_server_callback;
 
        using callbacks_t = utils::callback_manager<
            connection_callback_t,
            disconnection_callback_t,
            receive_callback_t,
            error_callback_t
        >;
 
        // =====================================================================
        // Member Variables
        // =====================================================================
 
        std::string server_id_;               
        // Note: lifecycle_ and stop_initiated_ are managed by startable_base
        callbacks_t callbacks_;               
        std::shared_ptr<asio::io_context>
            io_context_;    
        std::unique_ptr<asio::executor_work_guard<asio::io_context::executor_type>>
            work_guard_; 
        std::unique_ptr<asio::ip::tcp::acceptor>
            acceptor_;      
        std::future<void>
            io_context_future_; 
        std::vector<std::shared_ptr<kcenon::network::session::messaging_session>> sessions_;
 
        std::mutex sessions_mutex_;
 
        mutable std::mutex acceptor_mutex_;
 
        std::unique_ptr<asio::steady_timer> cleanup_timer_;
 
#if KCENON_WITH_COMMON_SYSTEM
        kcenon::common::interfaces::IMonitor* monitor_ = nullptr;
 
        std::atomic<uint64_t> messages_received_{0};
        std::atomic<uint64_t> messages_sent_{0};
        std::atomic<uint64_t> connection_errors_{0};
#endif // KCENON_WITH_COMMON_SYSTEM
    };
 
} // namespace kcenon::network::core
 
#endif // KCENON_NETWORK_INTERNAL_CORE_MESSAGING_SERVER_H_