transport.h

Go to the documentation of this file.

// SPDX-FileCopyrightText: Copyright (c) 2024 Cisco Systems
// SPDX-License-Identifier: BSD-2-Clause
 
#pragma once
 
#include "messages.h"
#include "tick_service.h"
 
#include "quic_transport.h"
 
#include "span.h"
#include <chrono>
#include <quicr/common.h>
#include <quicr/config.h>
#include <quicr/fetch_track_handler.h>
#include <quicr/metrics.h>
#include <quicr/publish_track_handler.h>
#include <quicr/subscribe_track_handler.h>
#include <spdlog/sinks/stdout_color_sinks.h>
#include <spdlog/spdlog.h>
 
#include <map>
#include <string>
#include <string_view>
 
namespace quicr {
 
    class Transport : public ITransport::TransportDelegate
    {
      public:
        Transport() = delete;
 
        enum class Status : uint8_t
        {
            kReady = 0,
            kNotReady,
 
            kInternalError,
 
            kInvalidParams,
 
            kConnecting,
            kDisconnecting,
            kNotConnected,
            kFailedToConnect,
            kPendingSeverSetup,
        };
 
        enum class ControlMessageStatus : uint8_t
        {
            kMessageIncomplete,        
            kMessageComplete,          
            kStreamBufferCannotBeZero, 
            kStreamBufferMissingType,  
            kUnsupportedMessageType,   
        };
 
        enum class StreamDataMessageStatus : uint8_t
        {
        };
 
        enum class ConnectionStatus : uint8_t
        {
            kNotConnected = 0,
            kConnecting,
            kConnected,
            kIdleTimeout,
            kClosedByRemote
        };
 
        struct ConnectionRemoteInfo
        {
            std::string ip; 
            uint16_t port;  
        };
 
        Transport(const ClientConfig& cfg, std::shared_ptr<TickService> tick_service);
 
        Transport(const ServerConfig& cfg, std::shared_ptr<TickService> tick_service);
 
        ~Transport();
 
        const std::shared_ptr<TickService>& GetTickService() const noexcept { return tick_service_; }
 
        // -------------------------------------------------------------------------------------------------
        // Public API MoQ Instance API methods
        // -------------------------------------------------------------------------------------------------
        void SubscribeTrack(ConnectionHandle connection_handle, std::shared_ptr<SubscribeTrackHandler> track_handler);
 
        void UnsubscribeTrack(ConnectionHandle connection_handle,
                              const std::shared_ptr<SubscribeTrackHandler>& track_handler);
 
        void UpdateTrackSubscription(ConnectionHandle connection_handle,
                                     std::shared_ptr<SubscribeTrackHandler> track_handler);
 
        void PublishTrack(ConnectionHandle connection_handle, std::shared_ptr<PublishTrackHandler> track_handler);
 
        void UnpublishTrack(ConnectionHandle connection_handle,
                            const std::shared_ptr<PublishTrackHandler>& track_handler);
 
        void FetchTrack(ConnectionHandle connection_handle, std::shared_ptr<FetchTrackHandler> track_handler);
 
        void CancelFetchTrack(ConnectionHandle connection_handle, std::shared_ptr<FetchTrackHandler> track_handler);
 
        Status GetStatus() const noexcept { return status_; }
 
 
        virtual void StatusChanged([[maybe_unused]] Status status) {}
 
      protected:
        Status Start();
 
        Status Stop();
 
 
      private:
        // -------------------------------------------------------------------------------------------------
        // Transport Delegate/callback functions
        // -------------------------------------------------------------------------------------------------
 
        void OnNewDataContext([[maybe_unused]] const ConnectionHandle& connection_handle,
                              [[maybe_unused]] const DataContextId& data_ctx_id) override;
 
        void OnConnectionStatus(const ConnectionHandle& connection_handle, const TransportStatus status) override;
        void OnNewConnection(const ConnectionHandle& connection_handle, const TransportRemote& remote) override;
        void OnRecvStream(const ConnectionHandle& connection_handle,
                          uint64_t stream_id,
                          std::optional<DataContextId> data_ctx_id,
                          const bool is_bidir = false) override;
        void OnRecvDgram(const ConnectionHandle& connection_handle, std::optional<DataContextId> data_ctx_id) override;
 
        void OnConnectionMetricsSampled(MetricsTimeStamp sample_time,
                                        TransportConnId conn_id,
                                        const QuicConnectionMetrics& quic_connection_metrics) override;
 
        void OnDataMetricsStampled(MetricsTimeStamp sample_time,
                                   TransportConnId conn_id,
                                   DataContextId data_ctx_id,
                                   const QuicDataContextMetrics& quic_data_context_metrics) override;
 
        // -------------------------------------------------------------------------------------------------
        // End of transport handler/callback functions
        // -------------------------------------------------------------------------------------------------
 
        static constexpr std::size_t kControlMessageBufferSize = 4096;
        struct ConnectionContext
        {
            ConnectionHandle connection_handle{ 0 };
            std::optional<uint64_t> ctrl_data_ctx_id;
            bool setup_complete{ false }; 
            uint64_t client_version{ 0 };
            std::optional<messages::ControlMessageType>
              ctrl_msg_type_received; 
 
            std::vector<uint8_t> ctrl_msg_buffer; 
 
            uint64_t current_subscribe_id{ 0 }; 
 
            std::map<messages::SubscribeId, std::pair<TrackNamespaceHash, TrackNameHash>> recv_sub_id;
 
            std::map<messages::SubscribeId, std::shared_ptr<SubscribeTrackHandler>> tracks_by_sub_id;
 
            std::map<messages::TrackAlias, std::shared_ptr<SubscribeTrackHandler>> sub_by_track_alias;
 
            std::map<TrackNamespaceHash, std::map<TrackNameHash, std::shared_ptr<PublishTrackHandler>>>
              pub_tracks_by_name;
 
            std::map<messages::TrackAlias, std::shared_ptr<PublishTrackHandler>> pub_tracks_by_track_alias;
 
            std::map<DataContextId, std::shared_ptr<PublishTrackHandler>> pub_tracks_by_data_ctx_id;
 
            ConnectionMetrics metrics{}; 
 
            ConnectionContext() { ctrl_msg_buffer.reserve(kControlMessageBufferSize); }
        };
 
        // -------------------------------------------------------------------------------------------------
        // Private methods
        // -------------------------------------------------------------------------------------------------
 
        void Init();
 
        PublishTrackHandler::PublishObjectStatus SendData(PublishTrackHandler& track_handler,
                                                          uint8_t priority,
                                                          uint32_t ttl,
                                                          bool stream_header_needed,
                                                          std::shared_ptr<const std::vector<uint8_t>> data);
 
        PublishTrackHandler::PublishObjectStatus SendObject(PublishTrackHandler& track_handler,
                                                            uint8_t priority,
                                                            uint32_t ttl,
                                                            bool stream_header_needed,
                                                            uint64_t group_id,
                                                            uint64_t subgroup_id,
                                                            uint64_t object_id,
                                                            std::optional<Extensions> extensions,
                                                            BytesSpan data);
 
        void SendCtrlMsg(const ConnectionContext& conn_ctx, BytesSpan data);
        void SendClientSetup();
        void SendServerSetup(ConnectionContext& conn_ctx);
        void SendAnnounce(ConnectionContext& conn_ctx, const TrackNamespace& track_namespace);
        void SendAnnounceOk(ConnectionContext& conn_ctx, const TrackNamespace& track_namespace);
        void SendUnannounce(ConnectionContext& conn_ctx, const TrackNamespace& track_namespace);
        void SendSubscribe(ConnectionContext& conn_ctx,
                           uint64_t subscribe_id,
                           const FullTrackName& tfn,
                           TrackHash th,
                           messages::ObjectPriority priority,
                           messages::GroupOrder group_order,
                           messages::FilterType filter_type);
        void SendSubscribeUpdate(ConnectionContext& conn_ctx,
                                 uint64_t subscribe_id,
                                 TrackHash th,
                                 messages::GroupId start_group_id,
                                 messages::ObjectId start_object_id,
                                 messages::GroupId end_group_id,
                                 messages::ObjectPriority priority);
 
        void SendSubscribeOk(ConnectionContext& conn_ctx,
                             uint64_t subscribe_id,
                             uint64_t expires,
                             bool content_exists,
                             messages::GroupId largest_group_id,
                             messages::ObjectId largest_object_id);
        void SendUnsubscribe(ConnectionContext& conn_ctx, uint64_t subscribe_id);
        void SendSubscribeDone(ConnectionContext& conn_ctx, uint64_t subscribe_id, const std::string& reason);
        void SendSubscribeError(ConnectionContext& conn_ctx,
                                uint64_t subscribe_id,
                                uint64_t track_alias,
                                messages::SubscribeErrorCode error,
                                const std::string& reason);
 
        void SendSubscribeAnnounces(ConnectionHandle conn_handle, const TrackNamespace& prefix_namespace);
        void SendUnsubscribeAnnounces(ConnectionHandle conn_handle, const TrackNamespace& prefix_namespace);
        void SendSubscribeAnnouncesOk(ConnectionContext& conn_ctx, const TrackNamespace& prefix_namespace);
        void SendSubscribeAnnouncesError(ConnectionContext& conn_ctx,
                                         const TrackNamespace& prefix_namespace,
                                         messages::SubscribeAnnouncesErrorCode err_code,
                                         const messages::ReasonPhrase& reason);
 
        void SendFetch(ConnectionContext& conn_ctx,
                       uint64_t subscribe_id,
                       const FullTrackName& tfn,
                       messages::ObjectPriority priority,
                       messages::GroupOrder group_order,
                       messages::GroupId start_group,
                       messages::GroupId start_object,
                       messages::GroupId end_group,
                       messages::GroupId end_object);
        void SendFetchCancel(ConnectionContext& conn_ctx, uint64_t subscribe_id);
        void SendFetchOk(ConnectionContext& conn_ctx,
                         uint64_t subscribe_id,
                         messages::GroupOrder group_order,
                         bool end_of_track,
                         messages::GroupId largest_group_id,
                         messages::GroupId largest_object_id);
        void SendFetchError(ConnectionContext& conn_ctx,
                            uint64_t subscribe_id,
                            messages::FetchErrorCode error,
                            const std::string& reason);
        void CloseConnection(ConnectionHandle connection_handle,
                             messages::TerminationReason reason,
                             const std::string& reason_str);
 
        void RemoveSubscribeTrack(ConnectionContext& conn_ctx,
                                  SubscribeTrackHandler& handler,
                                  bool remove_handler = true);
 
        void SendNewGroupRequest(ConnectionHandle conn_id, uint64_t subscribe_id, uint64_t track_alias);
 
        std::shared_ptr<PublishTrackHandler> GetPubTrackHandler(ConnectionContext& conn_ctx, TrackHash& th);
 
        void RemoveAllTracksForConnectionClose(ConnectionContext& conn_ctx);
 
        // -------------------------------------------------------------------------------------------------
        // Private member functions that will be implemented by Server class
        // -------------------------------------------------------------------------------------------------
        virtual void NewConnectionAccepted(ConnectionHandle, const ConnectionRemoteInfo&) {}
 
        virtual void ConnectionStatusChanged(ConnectionHandle, ConnectionStatus) {}
 
        virtual void SetConnectionHandle(ConnectionHandle) {}
 
        virtual void MetricsSampled(ConnectionHandle, const ConnectionMetrics&) {}
 
        // -------------------------------------------------------------------------------------------------
        // Private member functions that will be implemented by Client class
        // -------------------------------------------------------------------------------------------------
        virtual void MetricsSampled(const ConnectionMetrics&) {}
 
        // -------------------------------------------------------------------------------------------------
 
      private:
        // -------------------------------------------------------------------------------------------------
        // Private member functions that will be implemented by both Server and Client
        // ------------------------------------------------------------------------------------------------
 
        virtual bool ProcessCtrlMessage(ConnectionContext& conn_ctx, BytesSpan msg_bytes) = 0;
 
      private:
        // -------------------------------------------------------------------------------------------------
        // Private member variables
        // -------------------------------------------------------------------------------------------------
        std::mutex state_mutex_;
        const bool client_mode_;
        std::shared_ptr<spdlog::logger> logger_;
        bool stop_{ false };
        const ServerConfig server_config_;
        const ClientConfig client_config_;
 
        std::map<ConnectionHandle, ConnectionContext> connections_;
 
        Status status_{ Status::kNotReady };
 
        std::shared_ptr<TickService> tick_service_;
        std::shared_ptr<ITransport> quic_transport_; // **MUST** be last for proper order of destruction
 
        friend class Client;
        friend class Server;
    };
 
} // namespace quicr