stream_manager.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 "stream.h"
#include "kcenon/network/detail/utils/result_types.h"
 
#include <cstdint>
#include <functional>
#include <map>
#include <memory>
#include <mutex>
#include <shared_mutex>
#include <vector>
 
namespace kcenon::network::protocols::quic
{
 
class stream_manager
{
public:
    explicit stream_manager(bool is_server, uint64_t initial_max_stream_data = 65536);
 
    ~stream_manager() = default;
 
    // Non-copyable
    stream_manager(const stream_manager&) = delete;
    auto operator=(const stream_manager&) -> stream_manager& = delete;
 
    // Non-movable (due to mutex)
    stream_manager(stream_manager&&) = delete;
    auto operator=(stream_manager&&) -> stream_manager& = delete;
 
    // ========================================================================
    // Stream Creation
    // ========================================================================
 
    [[nodiscard]] auto create_bidirectional_stream() -> Result<uint64_t>;
 
    [[nodiscard]] auto create_unidirectional_stream() -> Result<uint64_t>;
 
    // ========================================================================
    // Stream Access
    // ========================================================================
 
    [[nodiscard]] auto get_stream(uint64_t stream_id) -> stream*;
 
    [[nodiscard]] auto get_stream(uint64_t stream_id) const -> const stream*;
 
    [[nodiscard]] auto get_or_create_stream(uint64_t stream_id) -> Result<stream*>;
 
    [[nodiscard]] auto has_stream(uint64_t stream_id) const -> bool;
 
    [[nodiscard]] auto stream_ids() const -> std::vector<uint64_t>;
 
    [[nodiscard]] auto stream_count() const -> size_t;
 
    // ========================================================================
    // Stream Limits (Transport Parameters)
    // ========================================================================
 
    void set_peer_max_streams_bidi(uint64_t max);
 
    void set_peer_max_streams_uni(uint64_t max);
 
    [[nodiscard]] auto local_max_streams_bidi() const noexcept -> uint64_t
    {
        return local_max_streams_bidi_;
    }
 
    [[nodiscard]] auto local_max_streams_uni() const noexcept -> uint64_t
    {
        return local_max_streams_uni_;
    }
 
    void set_local_max_streams_bidi(uint64_t max);
 
    void set_local_max_streams_uni(uint64_t max);
 
    [[nodiscard]] auto peer_max_streams_bidi() const noexcept -> uint64_t
    {
        return peer_max_streams_bidi_;
    }
 
    [[nodiscard]] auto peer_max_streams_uni() const noexcept -> uint64_t
    {
        return peer_max_streams_uni_;
    }
 
    // ========================================================================
    // Stream Queries
    // ========================================================================
 
    [[nodiscard]] auto streams_with_pending_data() -> std::vector<stream*>;
 
    [[nodiscard]] auto streams_needing_flow_control_update() -> std::vector<stream*>;
 
    void for_each_stream(const std::function<void(stream&)>& callback);
 
    void for_each_stream(const std::function<void(const stream&)>& callback) const;
 
    // ========================================================================
    // Stream Lifecycle
    // ========================================================================
 
    auto remove_closed_streams() -> size_t;
 
    void close_all_streams(uint64_t error_code);
 
    void reset();
 
    // ========================================================================
    // Statistics
    // ========================================================================
 
    [[nodiscard]] auto local_bidi_streams_count() const noexcept -> uint64_t
    {
        return next_local_bidi_id_ / 4;
    }
 
    [[nodiscard]] auto local_uni_streams_count() const noexcept -> uint64_t
    {
        return (next_local_uni_id_ - 2) / 4;
    }
 
    [[nodiscard]] auto peer_bidi_streams_count() const noexcept -> uint64_t
    {
        return (highest_peer_bidi_id_ + 4 - peer_bidi_type_) / 4;
    }
 
    [[nodiscard]] auto peer_uni_streams_count() const noexcept -> uint64_t
    {
        return (highest_peer_uni_id_ + 4 - peer_uni_type_) / 4;
    }
 
private:
    bool is_server_;
    uint64_t initial_max_stream_data_;
 
    // Stream ID generators
    // Client: bidi=0,4,8..., uni=2,6,10...
    // Server: bidi=1,5,9..., uni=3,7,11...
    uint64_t next_local_bidi_id_;
    uint64_t next_local_uni_id_;
 
    // Peer stream ID tracking
    uint64_t highest_peer_bidi_id_{0};
    uint64_t highest_peer_uni_id_{0};
    uint64_t peer_bidi_type_;
    uint64_t peer_uni_type_;
 
    // Stream limits
    uint64_t local_max_streams_bidi_{100};
    uint64_t local_max_streams_uni_{100};
    uint64_t peer_max_streams_bidi_{0};
    uint64_t peer_max_streams_uni_{0};
 
    // Active streams
    mutable std::shared_mutex streams_mutex_;
    std::map<uint64_t, std::unique_ptr<stream>> streams_;
 
    // Internal helpers
    [[nodiscard]] auto validate_stream_id(uint64_t stream_id) const -> VoidResult;
    [[nodiscard]] auto is_local_stream(uint64_t stream_id) const noexcept -> bool;
    [[nodiscard]] auto can_create_local_stream(bool bidirectional) const -> bool;
    [[nodiscard]] auto can_accept_peer_stream(uint64_t stream_id) const -> bool;
};
 
} // namespace kcenon::network::protocols::quic