http2_client.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 "frame.h"
#include "hpack.h"
#include "kcenon/network/detail/utils/result_types.h"
 
#include <asio.hpp>
#include <asio/ssl.hpp>
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <deque>
#include <functional>
#include <future>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <vector>
 
#if defined(NETWORK_ENABLE_TEST_INJECTION)
namespace kcenon::network::tests::support
{
    class http2_client_test_access;
} // namespace kcenon::network::tests::support
#endif
 
namespace kcenon::network::protocols::http2
{
    enum class stream_state
    {
        idle,              
        open,              
        half_closed_local, 
        half_closed_remote,
        closed             
    };
 
    struct http2_response
    {
        int status_code = 0;                              
        std::vector<http_header> headers;                 
        std::vector<uint8_t> body;                        
 
        auto get_header(const std::string& name) const -> std::optional<std::string>;
 
        auto get_body_string() const -> std::string;
    };
 
    struct http2_stream
    {
        uint32_t stream_id = 0;               
        std::atomic<stream_state> state{stream_state::idle};  
        std::vector<http_header> request_headers;  
        std::vector<http_header> response_headers; 
        std::vector<uint8_t> request_body;    
        std::vector<uint8_t> response_body;   
        int32_t window_size = 65535;          
        std::promise<http2_response> promise; 
        bool headers_complete = false;        
        bool body_complete = false;           
 
        bool is_streaming = false;            
        std::function<void(std::vector<uint8_t>)> on_data;  
        std::function<void(std::vector<http_header>)> on_headers;  
        std::function<void(int)> on_complete; 
 
        http2_stream() = default;
        http2_stream(http2_stream&& other) noexcept { *this = std::move(other); }
        auto operator=(http2_stream&& other) noexcept -> http2_stream&
        {
            if (this == &other) return *this;
            stream_id = other.stream_id;
            state.store(other.state.load());
            request_headers = std::move(other.request_headers);
            response_headers = std::move(other.response_headers);
            request_body = std::move(other.request_body);
            response_body = std::move(other.response_body);
            window_size = other.window_size;
            promise = std::move(other.promise);
            headers_complete = other.headers_complete;
            body_complete = other.body_complete;
            is_streaming = other.is_streaming;
            on_data = std::move(other.on_data);
            on_headers = std::move(other.on_headers);
            on_complete = std::move(other.on_complete);
            return *this;
        }
    };
 
    struct http2_settings
    {
        uint32_t header_table_size = 4096;        
        bool enable_push = false;                 
        uint32_t max_concurrent_streams = 100;    
        uint32_t initial_window_size = 65535;     
        uint32_t max_frame_size = 16384;          
        uint32_t max_header_list_size = 8192;     
    };
 
    class http2_client : public std::enable_shared_from_this<http2_client>
    {
    public:
        explicit http2_client(std::string_view client_id);
 
        ~http2_client();
 
        // Non-copyable
        http2_client(const http2_client&) = delete;
        http2_client& operator=(const http2_client&) = delete;
 
        // Non-movable (contains std::atomic members)
        http2_client(http2_client&&) = delete;
        http2_client& operator=(http2_client&&) = delete;
 
        auto connect(const std::string& host, unsigned short port = 443) -> VoidResult;
 
        auto disconnect() -> VoidResult;
 
        auto is_connected() const -> bool;
 
        auto get(const std::string& path,
                 const std::vector<http_header>& headers = {})
            -> Result<http2_response>;
 
        auto post(const std::string& path,
                  const std::string& body,
                  const std::vector<http_header>& headers = {})
            -> Result<http2_response>;
 
        auto post(const std::string& path,
                  const std::vector<uint8_t>& body,
                  const std::vector<http_header>& headers = {})
            -> Result<http2_response>;
 
        auto put(const std::string& path,
                 const std::string& body,
                 const std::vector<http_header>& headers = {})
            -> Result<http2_response>;
 
        auto del(const std::string& path,
                 const std::vector<http_header>& headers = {})
            -> Result<http2_response>;
 
        auto set_timeout(std::chrono::milliseconds timeout) -> void;
 
        auto get_timeout() const -> std::chrono::milliseconds;
 
        auto start_stream(const std::string& path,
                          const std::vector<http_header>& headers,
                          std::function<void(std::vector<uint8_t>)> on_data,
                          std::function<void(std::vector<http_header>)> on_headers,
                          std::function<void(int)> on_complete)
            -> Result<uint32_t>;
 
        auto write_stream(uint32_t stream_id,
                          const std::vector<uint8_t>& data,
                          bool end_stream = false) -> VoidResult;
 
        auto close_stream_writer(uint32_t stream_id) -> VoidResult;
 
        auto cancel_stream(uint32_t stream_id) -> VoidResult;
 
        auto get_settings() const -> http2_settings;
 
        auto set_settings(const http2_settings& settings) -> void;
 
    private:
        // Connection management
        auto send_connection_preface() -> VoidResult;
        auto send_settings() -> VoidResult;
        auto handle_settings_frame(const settings_frame& frame) -> VoidResult;
        auto send_settings_ack() -> VoidResult;
 
        // Frame I/O
        auto send_frame(const frame& f) -> VoidResult;
        auto read_next_frame() -> void;
        auto write_next_frame() -> void;
        auto process_frame(std::unique_ptr<frame> f) -> VoidResult;
 
        // Stream management
        auto allocate_stream_id() -> uint32_t;
        auto get_stream(uint32_t stream_id) -> http2_stream*;
        auto create_stream() -> http2_stream&;
        auto close_stream(uint32_t stream_id) -> void;
 
        // Request handling
        auto send_request(const std::string& method,
                          const std::string& path,
                          const std::vector<http_header>& headers,
                          const std::vector<uint8_t>& body)
            -> Result<http2_response>;
        auto build_headers(const std::string& method,
                           const std::string& path,
                           const std::vector<http_header>& additional) -> std::vector<http_header>;
 
        // Response handling
        auto handle_headers_frame(const headers_frame& f) -> VoidResult;
        auto handle_data_frame(const data_frame& f) -> VoidResult;
        auto handle_rst_stream_frame(const rst_stream_frame& f) -> VoidResult;
        auto handle_goaway_frame(const goaway_frame& f) -> VoidResult;
        auto handle_window_update_frame(const window_update_frame& f) -> VoidResult;
        auto handle_ping_frame(const ping_frame& f) -> VoidResult;
 
        // I/O thread
        auto run_io() -> void;
        auto stop_io() -> void;
 
        // Member variables
        std::string client_id_;                          
        std::string host_;                               
        unsigned short port_ = 443;                      
 
        // ASIO context
        // Serialize connection setup and teardown, including failed handshakes.
        std::mutex lifecycle_mutex_;
        std::unique_ptr<asio::io_context> io_context_;
        std::unique_ptr<asio::ssl::context> ssl_context_;
        std::unique_ptr<asio::ssl::stream<asio::ip::tcp::socket>> socket_;
        std::unique_ptr<asio::executor_work_guard<asio::io_context::executor_type>> work_guard_;
        std::future<void> io_future_;
        // The I/O thread owns every SSL operation after connection setup.
        // Protect submission against shutdown; queued buffers live through
        // asynchronous completion and writes never overlap one another.
        std::mutex io_submission_mutex_;
        bool io_active_ = false;
        struct pending_write
        {
            std::vector<uint8_t> data;
            std::shared_ptr<std::promise<VoidResult>> completion;
        };
        std::deque<pending_write> pending_writes_;
 
        // Connection state
        std::atomic<bool> is_connected_{false};
        std::atomic<bool> is_running_{false};
        std::atomic<bool> goaway_received_{false};
 
        // Stream management
        std::mutex streams_mutex_;
        std::map<uint32_t, http2_stream> streams_;
        std::atomic<uint32_t> next_stream_id_{1};        
        int32_t connection_window_size_ = 65535;
 
        // Settings (must be declared before encoder_/decoder_ for initialization order)
        http2_settings local_settings_;
        http2_settings remote_settings_;
 
        // HPACK (uses local_settings_.header_table_size in constructor)
        hpack_encoder encoder_;
        hpack_decoder decoder_;
 
        // Timeout
        std::atomic<std::chrono::milliseconds> timeout_{std::chrono::milliseconds(30000)};
 
        // Read buffer
        std::vector<uint8_t> read_buffer_;
 
        // Constants
        static constexpr std::string_view CONNECTION_PREFACE =
            "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
        static constexpr size_t FRAME_HEADER_SIZE = 9;
        static constexpr size_t DEFAULT_WINDOW_SIZE = 65535;
 
#if defined(NETWORK_ENABLE_TEST_INJECTION)
        // Test-only: grants tests/support/http2_client_test_access access to
        // the private process_frame() dispatcher and selected member fields
        // without leaking them through the public API. Used by
        // tests/unit/http2_client_branch_test.cpp to bypass the SETTINGS
        // handshake gate that exceeds 15 s under coverage instrumentation
        // (Issue #1115).
        friend class kcenon::network::tests::support::http2_client_test_access;
#endif
    };
 
} // namespace kcenon::network::protocols::http2