stream.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_types.h"
#include "kcenon/network/detail/utils/result_types.h"
 
#include <cstdint>
#include <deque>
#include <map>
#include <mutex>
#include <optional>
#include <span>
#include <vector>
 
namespace kcenon::network::protocols::quic
{
 
namespace stream_id_type
{
    constexpr uint64_t client_bidi = 0x00;
    constexpr uint64_t server_bidi = 0x01;
    constexpr uint64_t client_uni = 0x02;
    constexpr uint64_t server_uni = 0x03;
 
    [[nodiscard]] constexpr auto is_client_initiated(uint64_t stream_id) noexcept -> bool
    {
        return (stream_id & 0x01) == 0;
    }
 
    [[nodiscard]] constexpr auto is_server_initiated(uint64_t stream_id) noexcept -> bool
    {
        return (stream_id & 0x01) == 1;
    }
 
    [[nodiscard]] constexpr auto is_bidirectional(uint64_t stream_id) noexcept -> bool
    {
        return (stream_id & 0x02) == 0;
    }
 
    [[nodiscard]] constexpr auto is_unidirectional(uint64_t stream_id) noexcept -> bool
    {
        return (stream_id & 0x02) != 0;
    }
 
    [[nodiscard]] constexpr auto get_type(uint64_t stream_id) noexcept -> uint64_t
    {
        return stream_id & 0x03;
    }
 
    [[nodiscard]] constexpr auto get_sequence(uint64_t stream_id) noexcept -> uint64_t
    {
        return stream_id >> 2;
    }
 
    [[nodiscard]] constexpr auto make_stream_id(uint64_t type, uint64_t sequence) noexcept -> uint64_t
    {
        return (sequence << 2) | (type & 0x03);
    }
} // namespace stream_id_type
 
enum class send_stream_state
{
    ready,       
    send,        
    data_sent,   
    reset_sent,  
    reset_recvd, 
    data_recvd,  
};
 
enum class recv_stream_state
{
    recv,        
    size_known,  
    data_recvd,  
    reset_recvd, 
    data_read,   
    reset_read,  
};
 
namespace stream_error
{
    constexpr int invalid_stream_id = -700;
    constexpr int stream_not_found = -701;
    constexpr int stream_limit_exceeded = -702;
    constexpr int flow_control_error = -703;
    constexpr int final_size_error = -704;
    constexpr int stream_state_error = -705;
    constexpr int stream_reset = -706;
    constexpr int buffer_full = -707;
} // namespace stream_error
 
class stream
{
public:
    explicit stream(uint64_t id, bool is_local, uint64_t initial_max_data = 65536);
 
    ~stream() = default;
 
    // Non-copyable
    stream(const stream&) = delete;
    auto operator=(const stream&) -> stream& = delete;
 
    // Movable
    stream(stream&&) noexcept = default;
    auto operator=(stream&&) noexcept -> stream& = default;
 
    // ========================================================================
    // Stream Properties
    // ========================================================================
 
    [[nodiscard]] auto id() const noexcept -> uint64_t { return id_; }
 
    [[nodiscard]] auto is_local() const noexcept -> bool { return is_local_; }
 
    [[nodiscard]] auto is_unidirectional() const noexcept -> bool
    {
        return stream_id_type::is_unidirectional(id_);
    }
 
    [[nodiscard]] auto is_bidirectional() const noexcept -> bool
    {
        return stream_id_type::is_bidirectional(id_);
    }
 
    // ========================================================================
    // Send Side
    // ========================================================================
 
    [[nodiscard]] auto send_state() const noexcept -> send_stream_state { return send_state_; }
 
    [[nodiscard]] auto can_send() const noexcept -> bool;
 
    [[nodiscard]] auto write(std::span<const uint8_t> data) -> Result<size_t>;
 
    [[nodiscard]] auto finish() -> VoidResult;
 
    [[nodiscard]] auto reset(uint64_t error_code) -> VoidResult;
 
    [[nodiscard]] auto fin_sent() const noexcept -> bool { return fin_sent_; }
 
    [[nodiscard]] auto pending_bytes() const noexcept -> size_t { return send_buffer_.size(); }
 
    [[nodiscard]] auto next_stream_frame(size_t max_size) -> std::optional<stream_frame>;
 
    void acknowledge_data(uint64_t offset, uint64_t length);
 
    // ========================================================================
    // Receive Side
    // ========================================================================
 
    [[nodiscard]] auto recv_state() const noexcept -> recv_stream_state { return recv_state_; }
 
    [[nodiscard]] auto has_data() const noexcept -> bool;
 
    [[nodiscard]] auto read(std::span<uint8_t> buffer) -> Result<size_t>;
 
    [[nodiscard]] auto is_fin_received() const noexcept -> bool { return recv_fin_; }
 
    [[nodiscard]] auto stop_sending(uint64_t error_code) -> VoidResult;
 
    [[nodiscard]] auto receive_data(uint64_t offset, std::span<const uint8_t> data, bool fin)
        -> VoidResult;
 
    [[nodiscard]] auto receive_reset(uint64_t error_code, uint64_t final_size) -> VoidResult;
 
    [[nodiscard]] auto receive_stop_sending(uint64_t error_code) -> VoidResult;
 
    // ========================================================================
    // Flow Control
    // ========================================================================
 
    void set_max_send_data(uint64_t max);
 
    [[nodiscard]] auto max_send_data() const noexcept -> uint64_t { return max_send_offset_; }
 
    [[nodiscard]] auto available_send_window() const noexcept -> size_t;
 
    void set_max_recv_data(uint64_t max);
 
    [[nodiscard]] auto max_recv_data() const noexcept -> uint64_t { return max_recv_offset_; }
 
    [[nodiscard]] auto bytes_consumed() const noexcept -> uint64_t { return recv_offset_; }
 
    [[nodiscard]] auto should_send_max_stream_data() const noexcept -> bool;
 
    [[nodiscard]] auto generate_max_stream_data() -> std::optional<uint64_t>;
 
    // ========================================================================
    // Error Information
    // ========================================================================
 
    [[nodiscard]] auto reset_error_code() const noexcept -> std::optional<uint64_t>
    {
        return reset_error_code_;
    }
 
    [[nodiscard]] auto stop_sending_error_code() const noexcept -> std::optional<uint64_t>
    {
        return stop_sending_error_code_;
    }
 
private:
    uint64_t id_;
    bool is_local_;
 
    // Send state
    send_stream_state send_state_{send_stream_state::ready};
    std::deque<uint8_t> send_buffer_;
    uint64_t send_offset_{0};    // Next byte to send
    uint64_t acked_offset_{0};   // Highest contiguously acknowledged
    bool fin_sent_{false};
    bool fin_acked_{false};
 
    // Receive state
    recv_stream_state recv_state_{recv_stream_state::recv};
    std::map<uint64_t, std::vector<uint8_t>> recv_buffer_; // offset -> data (for gaps)
    std::deque<uint8_t> recv_ready_;  // Contiguous data ready for reading
    uint64_t recv_offset_{0};         // Next expected offset
    bool recv_fin_{false};
    std::optional<uint64_t> final_size_;
 
    // Flow control - send
    uint64_t max_send_offset_{0};     // Peer's MAX_STREAM_DATA
 
    // Flow control - receive
    uint64_t max_recv_offset_{65536}; // Our MAX_STREAM_DATA
    uint64_t recv_window_size_{65536};
    static constexpr double window_update_threshold_{0.5};
 
    // Error codes
    std::optional<uint64_t> reset_error_code_;
    std::optional<uint64_t> stop_sending_error_code_;
 
    // Internal helpers
    void reassemble_data();
    void update_send_state();
    void update_recv_state();
};
 
[[nodiscard]] auto send_state_to_string(send_stream_state state) -> const char*;
 
[[nodiscard]] auto recv_state_to_string(recv_stream_state state) -> const char*;
 
} // namespace kcenon::network::protocols::quic