messaging_session.cpp

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.
 
#include "kcenon/network/detail/session/messaging_session.h"
 
#include <span>
 
#include "internal/tcp/tcp_socket.h"
#include "internal/integration/logger_integration.h"
#include "kcenon/network/detail/metrics/network_metrics.h"
 
// Use nested namespace definition (C++17)
namespace kcenon::network::session
{
 
    // Use string_view in constructor for efficiency (C++17)
    messaging_session::messaging_session(asio::ip::tcp::socket socket,
                                         std::string_view server_id)
        : server_id_(server_id)
    {
        // Create the tcp_socket wrapper
        socket_ = std::make_shared<internal::tcp_socket>(std::move(socket));
    }
 
    messaging_session::~messaging_session() noexcept
    {
        try
        {
            // Call stop_session to clean up resources
            stop_session();
        }
        catch (...)
        {
            // Destructor must not throw - swallow all exceptions
        }
    }
 
    auto messaging_session::start_session() -> void
    {
        // NOTE: No logging in session methods to prevent heap corruption during
        // static destruction. These methods may run in async handlers when
        // GlobalLoggerRegistry is already destroyed.
 
        if (is_stopped_.load())
        {
            return;
        }
 
        // Set up callbacks with weak_ptr to avoid circular reference
        // Use span-based callback for zero-copy receive path (no per-read allocation)
        auto weak_self = weak_from_this();
        socket_->set_receive_callback_view(
            [weak_self](std::span<const uint8_t> data)
            {
                if (auto self = weak_self.lock())
                {
                    self->on_receive(data);
                }
            });
        socket_->set_error_callback([weak_self](std::error_code ec)
            {
                if (auto self = weak_self.lock())
                {
                    self->on_error(ec);
                }
            });
 
        // Begin reading
        socket_->start_read();
    }
 
    auto messaging_session::stop_session() -> void
    {
        if (is_stopped_.exchange(true))
        {
            return;
        }
        // Close socket safely using atomic close() method
        // This prevents data races between close and async read operations
        if (socket_)
        {
            socket_->close();
        }
 
        // Invoke disconnection callback if set
        {
            std::lock_guard<std::mutex> lock(callback_mutex_);
            if (disconnection_callback_)
            {
                try
                {
                    disconnection_callback_(server_id_);
                }
                catch (...)
                {
                    // Silently ignore exceptions - no logging during potential static destruction
                }
            }
        }
    }
 
    auto messaging_session::send_packet(std::vector<uint8_t>&& data) -> void
    {
        if (is_stopped_.load())
        {
            return;
        }
 
        // Report bytes sent metric
        metrics::metric_reporter::report_bytes_sent(data.size());
 
        // Send data directly without pipeline transformation
        // NOTE: No logging in async callbacks to prevent heap corruption during static destruction
        socket_->async_send(std::move(data),
            [](std::error_code, std::size_t) {
                // Silently ignore errors - no logging during potential static destruction
            });
    }
 
    auto messaging_session::send(std::vector<uint8_t>&& data) -> VoidResult
    {
        if (is_stopped_.load())
        {
            return error_void(
                error_codes::network_system::connection_closed,
                "Session is closed",
                "messaging_session");
        }
 
        if (data.empty())
        {
            return error_void(
                error_codes::common_errors::invalid_argument,
                "Data cannot be empty",
                "messaging_session");
        }
 
        // Delegate to send_packet for actual sending
        send_packet(std::move(data));
 
        return ok();
    }
 
    auto messaging_session::on_receive(std::span<const uint8_t> data) -> void
    {
        // NOTE: No logging in async handlers to prevent heap corruption during
        // static destruction.
 
        if (is_stopped_.load())
        {
            return;
        }
 
        // Report bytes received metric
        metrics::metric_reporter::report_bytes_received(data.size());
 
        // Check queue size before adding
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            size_t queue_size = pending_messages_.size();
 
            // Apply backpressure if queue is getting full
            // If queue is severely overflowing, disconnect the session
            if (queue_size >= max_pending_messages_ * 2)
            {
                stop_session();
                return;
            }
 
            // Copy span data into vector only when adding to queue (single copy point)
            // This is the only allocation in the receive path
            pending_messages_.emplace_back(data.begin(), data.end());
        }
 
        // Process messages from queue
        process_next_message();
    }
 
    auto messaging_session::on_error(std::error_code ec) -> void
    {
        // NOTE: No logging in async handlers to prevent heap corruption during
        // static destruction.
 
        // Invoke error callback if set
        {
            std::lock_guard<std::mutex> lock(callback_mutex_);
            if (error_callback_)
            {
                try
                {
                    error_callback_(ec);
                }
                catch (...)
                {
                    // Silently ignore exceptions - no logging during potential static destruction
                }
            }
        }
 
        stop_session();
    }
 
    auto messaging_session::process_next_message() -> void
    {
        // NOTE: No logging in message processing to prevent heap corruption during
        // static destruction.
 
        std::vector<uint8_t> message;
 
        // Dequeue one message
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            if (pending_messages_.empty())
            {
                return;
            }
 
            message = std::move(pending_messages_.front());
            pending_messages_.pop_front();
        }
 
        // Invoke receive callback if set
        {
            std::lock_guard<std::mutex> lock(callback_mutex_);
            if (receive_callback_)
            {
                try
                {
                    receive_callback_(message);
                }
                catch (...)
                {
                    // Silently ignore exceptions - no logging during potential static destruction
                }
            }
        }
 
        // If there are more messages, process them asynchronously
        // to avoid blocking the receive thread
        std::lock_guard<std::mutex> lock(queue_mutex_);
        if (!pending_messages_.empty())
        {
            // In a production system, you might want to post this to a thread pool
            // or use a work queue to process messages asynchronously
            // For now, we just process one message per call
        }
    }
 
    auto messaging_session::set_receive_callback(
        std::function<void(const std::vector<uint8_t>&)> callback) -> void
    {
        std::lock_guard<std::mutex> lock(callback_mutex_);
        receive_callback_ = std::move(callback);
    }
 
    auto messaging_session::set_disconnection_callback(
        std::function<void(const std::string&)> callback) -> void
    {
        std::lock_guard<std::mutex> lock(callback_mutex_);
        disconnection_callback_ = std::move(callback);
    }
 
    auto messaging_session::set_error_callback(
        std::function<void(std::error_code)> callback) -> void
    {
        std::lock_guard<std::mutex> lock(callback_mutex_);
        error_callback_ = std::move(callback);
    }
 
} // namespace kcenon::network::session