tcp_socket.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
 
#include <array>
#include <atomic>
#include <functional>
#include <memory>
#include <mutex>
#include <span>
#include <system_error>
#include <vector>
 
#include <asio.hpp>
 
#include "internal/utils/common_defs.h"
#include "kcenon/network-core/interfaces/socket_observer.h"
 
namespace kcenon::network::internal
{
    class tcp_socket : public std::enable_shared_from_this<tcp_socket>
    {
    public:
        using backpressure_callback = std::function<void(bool apply_backpressure)>;
 
        tcp_socket(asio::ip::tcp::socket socket);
 
        tcp_socket(asio::ip::tcp::socket socket, const socket_config& config);
 
        ~tcp_socket() = default;
 
        auto attach_observer(std::shared_ptr<network_core::interfaces::socket_observer> observer)
            -> void;
 
        auto detach_observer(std::shared_ptr<network_core::interfaces::socket_observer> observer)
            -> void;
 
        auto set_receive_callback(
            std::function<void(const std::vector<uint8_t>&)> callback) -> void;
 
        auto set_receive_callback_view(
            std::function<void(std::span<const uint8_t>)> callback) -> void;
 
        auto set_error_callback(std::function<void(std::error_code)> callback)
            -> void;
 
        auto start_read() -> void;
 
        auto async_send(
            std::vector<uint8_t>&& data,
            std::function<void(std::error_code, std::size_t)> handler) -> void;
 
        auto set_backpressure_callback(backpressure_callback callback) -> void;
 
        [[nodiscard]] auto try_send(
            std::vector<uint8_t>&& data,
            std::function<void(std::error_code, std::size_t)> handler) -> bool;
 
        [[nodiscard]] auto pending_bytes() const -> std::size_t;
 
        [[nodiscard]] auto is_backpressure_active() const -> bool;
 
        [[nodiscard]] auto metrics() const -> const socket_metrics&;
 
        auto reset_metrics() -> void;
 
        [[nodiscard]] auto config() const -> const socket_config&;
 
        auto socket() -> asio::ip::tcp::socket& { return socket_; }
 
        auto stop_read() -> void;
 
        auto close() -> void;
 
        [[nodiscard]] auto is_closed() const -> bool;
 
    private:
        auto do_read() -> void;
 
    private:
        using receive_callback_t = std::function<void(const std::vector<uint8_t>&)>;
        using receive_callback_view_t = std::function<void(std::span<const uint8_t>)>;
        using error_callback_t = std::function<void(std::error_code)>;
 
        asio::ip::tcp::socket socket_; 
        std::array<uint8_t, 4096>
            read_buffer_; 
        std::shared_ptr<receive_callback_t> receive_callback_;
        std::shared_ptr<receive_callback_view_t> receive_callback_view_;
        std::shared_ptr<error_callback_t> error_callback_;
 
        std::mutex callback_mutex_; 
        socket_config config_;
 
        mutable socket_metrics metrics_;
 
        std::vector<std::weak_ptr<network_core::interfaces::socket_observer>> observers_{};
 
        std::atomic<bool> is_reading_{false}; 
        std::atomic<bool> is_closed_{false};  
        std::atomic<std::size_t> pending_bytes_{0};
 
        std::atomic<bool> backpressure_active_{false};
 
        std::shared_ptr<backpressure_callback> backpressure_callback_;
 
        auto notify_observers_receive(std::span<const uint8_t> data) -> void;
 
        auto notify_observers_error(std::error_code ec) -> void;
 
        auto notify_observers_backpressure(bool apply) -> void;
    };
} // namespace kcenon::network::internal