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 "attributes.h"
#include "messages.h"
#include "tick_service.h"
 
#include "quic_transport.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 <span>
#include <spdlog/sinks/stdout_color_sinks.h>
#include <spdlog/spdlog.h>
 
#include <atomic>
#include <map>
#include <string>
#include <string_view>
 
namespace quicr {

    class Transport
      : public ITransport::TransportDelegate
      , public std::enable_shared_from_this<Transport>
    {
      public:
        Transport() = delete;

        enum class Status : uint8_t
        {
            kReady = 0,
            kNotReady,
 
            kInternalError,
 
            kInvalidParams,
 
            kConnecting,
            kDisconnecting,
            kNotConnected,
            kFailedToConnect,
            kPendingServerSetup,
        };

        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,
                                     bool new_group_request = false);

        void PublishTrack(ConnectionHandle connection_handle, std::shared_ptr<PublishTrackHandler> track_handler);

        void PublishTrackSub(ConnectionHandle connection_handle, std::shared_ptr<PublishTrackHandler> track_handler);

        void UnpublishTrack(ConnectionHandle connection_handle,
                            const std::shared_ptr<PublishTrackHandler>& track_handler);

        virtual bool FetchReceived(ConnectionHandle connection_handle,
                                   uint64_t request_id,
                                   const FullTrackName& track_full_name,
                                   const quicr::messages::FetchAttributes& attributes);

        void FetchTrack(ConnectionHandle connection_handle, std::shared_ptr<FetchTrackHandler> track_handler);

        void CancelFetchTrack(ConnectionHandle connection_handle, std::shared_ptr<FetchTrackHandler> track_handler);

        uint64_t RequestTrackStatus(ConnectionHandle connection_handle,
                                    const FullTrackName& track_full_name,
                                    const messages::SubscribeAttributes& subscribe_attributes);

        virtual void ResolveTrackStatus(ConnectionHandle connection_handle,
                                        uint64_t request_id,
                                        uint64_t track_alias,
                                        const SubscribeResponse& subscribe_response);

        Status GetStatus() const noexcept { return status_; }


        virtual void StatusChanged([[maybe_unused]] Status status) {}

        virtual void TrackStatusReceived(ConnectionHandle connection_handle,
                                         uint64_t request_id,
                                         const FullTrackName& track_full_name,
                                         const messages::SubscribeAttributes& subscribe_attributes);

        virtual void TrackStatusResponseReceived(ConnectionHandle connection_handle,
                                                 uint64_t request_id,
                                                 const SubscribeResponse& response);


      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
        {
            ConnectionContext(const ConnectionContext& other)
              : next_request_id(other.next_request_id.load(std::memory_order_seq_cst))
            {
            }
 
            ConnectionHandle connection_handle{ 0 };
            std::optional<uint64_t> ctrl_data_ctx_id;
            bool setup_complete{ false }; 
            bool closed{ false };
            uint64_t client_version{ 0 };
            std::optional<messages::ControlMessageType>
              ctrl_msg_type_received; 
 
            std::vector<uint8_t> ctrl_msg_buffer; 

            std::atomic<uint64_t> next_request_id{ 0 }; 

            struct SubscribeContext
            {
                FullTrackName track_full_name;
                TrackHash track_hash{ 0, 0 };
                std::optional<messages::Location> largest_location{ std::nullopt };
            };
 
            std::map<messages::RequestID, SubscribeContext> recv_req_id;

            std::map<messages::RequestID, std::shared_ptr<SubscribeTrackHandler>> sub_tracks_by_request_id;

            std::map<messages::TrackAlias, std::shared_ptr<SubscribeTrackHandler>> sub_by_recv_track_alias;

            std::map<TrackNamespaceHash, std::map<TrackNameHash, std::shared_ptr<PublishTrackHandler>>>
              pub_tracks_by_name;

            std::map<messages::TrackAlias, std::map<uint64_t, std::shared_ptr<PublishTrackHandler>>>
              pub_tracks_by_track_alias;

            std::map<messages::RequestID, TrackNamespaceHash> pub_tracks_ns_by_request_id;

            std::map<messages::RequestID, std::shared_ptr<PublishTrackHandler>> pub_tracks_by_request_id;

            std::map<DataContextId, std::shared_ptr<PublishTrackHandler>> pub_tracks_by_data_ctx_id;

            std::map<messages::RequestID, std::shared_ptr<PublishTrackHandler>> pub_fetch_tracks_by_sub_id;

            std::map<messages::RequestID, TrackNamespace> sub_announces_by_request_id;
 
            ConnectionMetrics metrics{}; 
 
            ConnectionContext() { ctrl_msg_buffer.reserve(kControlMessageBufferSize); }
 
            /*
             * Get the next request Id to use
             */
            uint64_t GetNextRequestId()
            {
                uint64_t rid = next_request_id;
                next_request_id += 2;
 
                return rid;
            }
        };
 
        // -------------------------------------------------------------------------------------------------
        // Private methods
        // -------------------------------------------------------------------------------------------------
 
        void Init();
 
        void SendCtrlMsg(const ConnectionContext& conn_ctx, BytesSpan data);
        void SendClientSetup();
        void SendServerSetup(ConnectionContext& conn_ctx);
        void SendAnnounce(ConnectionContext& conn_ctx,
                          messages::RequestID request_id,
                          const TrackNamespace& track_namespace);
        void SendAnnounceOk(ConnectionContext& conn_ctx, messages::RequestID request_id);
        void SendUnannounce(ConnectionContext& conn_ctx, const TrackNamespace& track_namespace);
        void SendSubscribe(ConnectionContext& conn_ctx,
                           messages::RequestID request_id,
                           const FullTrackName& tfn,
                           TrackHash th,
                           messages::SubscriberPriority priority,
                           messages::GroupOrder group_order,
                           messages::FilterType filter_type,
                           std::chrono::milliseconds delivery_timeout);
        void SendSubscribeUpdate(const ConnectionContext& conn_ctx,
                                 messages::RequestID request_id,
                                 messages::RequestID subscribe_request_id,
                                 TrackHash th,
                                 messages::Location start_location,
                                 messages::GroupId end_group_id,
                                 messages::SubscriberPriority priority,
                                 bool forward,
                                 bool new_group_request = false);
 
        void SendSubscribeOk(ConnectionContext& conn_ctx,
                             messages::RequestID request_id,
                             uint64_t track_alias,
                             uint64_t expires,
                             const std::optional<messages::Location>& largest_location);
        void SendUnsubscribe(ConnectionContext& conn_ctx, messages::RequestID request_id);
        void SendSubscribeDone(ConnectionContext& conn_ctx, messages::RequestID request_id, const std::string& reason);
        void SendSubscribeError(ConnectionContext& conn_ctx,
                                messages::RequestID request_id,
                                messages::SubscribeErrorCode error,
                                const std::string& reason);
 
        void SendTrackStatus(ConnectionContext& conn_ctx, messages::RequestID request_id, const FullTrackName& tfn);
        void SendTrackStatusOk(ConnectionContext& conn_ctx,
                               messages::RequestID request_id,
                               uint64_t track_alias,
                               uint64_t expires,
                               const std::optional<messages::Location>& largest_location);
        void SendTrackStatusError(ConnectionContext& conn_ctx,
                                  messages::RequestID request_id,
                                  messages::SubscribeErrorErrorCode error,
                                  const std::string& reason);
 
        void SendPublish(ConnectionContext& conn_ctx,
                         messages::RequestID request_id,
                         const FullTrackName& tfn,
                         uint64_t track_alias,
                         messages::GroupOrder group_order,
                         std::optional<messages::Location> largest_location,
                         bool forward);
 
        void SendPublishOk(ConnectionContext& conn_ctx,
                           messages::RequestID request_id,
                           bool forward,
                           messages::SubscriberPriority priority,
                           messages::GroupOrder group_order,
                           messages::FilterType filter_type);
 
        void SendPublishError(ConnectionContext& conn_ctx,
                              messages::RequestID request_id,
                              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, messages::RequestID request_id);
        void SendSubscribeAnnouncesError(ConnectionContext& conn_ctx,
                                         messages::RequestID request_id,
                                         messages::SubscribeNamespaceErrorCode err_code,
                                         const messages::ReasonPhrase& reason);
 
        void SendFetch(ConnectionContext& conn_ctx,
                       messages::RequestID request_id,
                       const FullTrackName& tfn,
                       messages::SubscriberPriority priority,
                       messages::GroupOrder group_order,
                       messages::GroupId start_group,
                       messages::GroupId start_object,
                       messages::GroupId end_group,
                       messages::GroupId end_object);
        void SendJoiningFetch(ConnectionContext& conn_ctx,
                              messages::RequestID request_id,
                              messages::SubscriberPriority priority,
                              messages::GroupOrder group_order,
                              messages::RequestID joining_request_id,
                              messages::GroupId joining_start,
                              bool absolute,
                              const messages::Parameters parameters);
        void SendFetchCancel(ConnectionContext& conn_ctx, messages::RequestID request_id);
        void SendFetchOk(ConnectionContext& conn_ctx,
                         messages::RequestID request_id,
                         messages::GroupOrder group_order,
                         bool end_of_track,
                         messages::Location end_location);
        void SendFetchError(ConnectionContext& conn_ctx,
                            messages::RequestID request_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);
 
        std::shared_ptr<PublishTrackHandler> GetPubTrackHandler(ConnectionContext& conn_ctx, TrackHash& th);
 
        void RemoveAllTracksForConnectionClose(ConnectionContext& conn_ctx);
 
        uint64_t GetNextRequestId();
 
        bool OnRecvSubgroup(messages::StreamHeaderType type,
                            std::vector<uint8_t>::const_iterator cursor_it,
                            StreamRxContext& rx_ctx,
                            std::uint64_t stream_id,
                            ConnectionContext& conn_ctx,
                            std::shared_ptr<const std::vector<uint8_t>> data) const;
        bool OnRecvFetch(std::vector<uint8_t>::const_iterator cursor_it,
                         StreamRxContext& rx_ctx,
                         std::uint64_t stream_id,
                         ConnectionContext& conn_ctx,
                         std::shared_ptr<const std::vector<uint8_t>> data) const;
 
        // -------------------------------------------------------------------------------------------------
        // 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&) {}
 
      protected:
        std::shared_ptr<Transport> GetSharedPtr();
 
        ConnectionContext& GetConnectionContext(ConnectionHandle conn);
 
        // -------------------------------------------------------------------------------------------------
 
      private:
        // -------------------------------------------------------------------------------------------------
        // Private member functions that will be implemented by both Server and Client
        // ------------------------------------------------------------------------------------------------
 
        virtual bool ProcessCtrlMessage(ConnectionContext& conn_ctx, BytesSpan msg_bytes) = 0;
 
        TransportError Enqueue(const TransportConnId& conn_id,
                               const DataContextId& data_ctx_id,
                               std::uint64_t group_id,
                               std::shared_ptr<const std::vector<uint8_t>> bytes,
                               const uint8_t priority,
                               const uint32_t ttl_ms,
                               const uint32_t delay_ms,
                               const ITransport::EnqueueFlags flags);
 
      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;
        friend class PublishTrackHandler;
        friend class PublishFetchHandler;
        friend class SubscribeTrackHandler;
    };
 
} // namespace quicr