session.h

Go to the documentation of this file.

// SPDX-FileCopyrightText: Copyright (c) 2024 Cisco Systems
// SPDX-License-Identifier: BSD-2-Clause
 
#pragma once
 
#include "quicr/attributes.h"
#include "quicr/config.h"
#include "quicr/connection.h"
#include "quicr/containers/stream_buffer.h"
#include "quicr/errors.h"
#include "quicr/handlers/fetch_track_handler.h"
#include "quicr/handlers/forwarding_subscribe_track_handler.h"
#include "quicr/handlers/publish_fetch_handler.h"
#include "quicr/handlers/publish_namespace_handler.h"
#include "quicr/handlers/publish_track_handler.h"
#include "quicr/handlers/subscribe_namespace_handler.h"
#include "quicr/handlers/subscribe_track_handler.h"
#include "quicr/handlers/track_status_handler.h"
#include "quicr/messages/message.h"
#include "quicr/messages/message_serialisation.h"
#include "quicr/metrics.h"
#include "quicr/transport.h"
 
#include <atomic>
#include <chrono>
#include <map>
#include <mutex>
#include <span>
#include <string>
#include <string_view>
 
namespace quicr {
 
    class Logger;
 
    struct SubscribeResponse : TrackStatusResponse
    {
        bool is_publisher_initiated = false; 
 
        std::optional<std::chrono::milliseconds> expires{};
    };
 
    struct FetchResponse
    {
        std::optional<messages::Location> largest_location = std::nullopt;
 
        messages::GroupOrder publisher_default_group_order =
          messages::GroupOrder::kAscending; 
    };
 
    class Session
      : public Connection::Delegate
      , public std::enable_shared_from_this<Session>
    {
        static constexpr std::size_t kControlMessageBufferSize = 4096;
 
      public:
        enum class Status : uint8_t
        {
            kReady = 0,        
            kNotReady,         
            kInternalError,    
            kInvalidParams,    
            kConnecting,       
            kDisconnecting,    
            kNotConnected,     
            kFailedToConnect,  
            kPendingPeerSetup, 
        };
 
        struct Callbacks;
        struct ClientCallbacks;
        struct ServerCallbacks;
 
        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;  
        };
 
        struct SubscribeContext
        {
            FullTrackName track_full_name; 
 
            TrackHash track_hash{ 0, 0 }; 
 
            std::optional<messages::Location> largest_location{ std::nullopt };
 
            std::shared_ptr<Stream> stream; 
        };
 
      public:
        static std::shared_ptr<Session> Create(const ClientConfig& cfg,
                                               std::shared_ptr<Transport> transport,
                                               std::shared_ptr<Connection> connection,
                                               std::shared_ptr<ClientCallbacks> callbacks,
                                               std::shared_ptr<Clock> clock,
                                               std::shared_ptr<Logger> logger = nullptr)
        {
            return std::shared_ptr<Session>(new Session(cfg,
                                                        std::move(transport),
                                                        std::move(connection),
                                                        std::move(callbacks),
                                                        std::move(clock),
                                                        std::move(logger)));
        }
 
        static std::shared_ptr<Session> Create(const ServerConfig& cfg,
                                               std::shared_ptr<Transport> transport,
                                               std::shared_ptr<Connection> connection,
                                               std::shared_ptr<ServerCallbacks> callbacks,
                                               std::shared_ptr<Clock> clock,
                                               std::shared_ptr<Logger> logger = nullptr)
        {
            return std::shared_ptr<Session>(new Session(cfg,
                                                        std::move(transport),
                                                        std::move(connection),
                                                        std::move(callbacks),
                                                        std::move(clock),
                                                        std::move(logger)));
        }
 
        virtual ~Session();
 
        std::shared_ptr<Connection> GetDefaultConnection() const noexcept;
 
        const std::shared_ptr<Clock>& GetTickService() const noexcept { return clock_; }
 
        const std::shared_ptr<Logger>& GetLogger() const noexcept { return logger_; }
 
        Status GetStatus() const noexcept { return status_.load(std::memory_order_acquire); }
 
        void Disconnect();
 
        /*===================================================================*/
        // Public API MoQ Instance API methods
        /*===================================================================*/
 
        void SubscribeTrack(std::shared_ptr<SubscribeTrackHandler> track_handler);
 
        void UnsubscribeTrack(const std::shared_ptr<SubscribeTrackHandler>& track_handler);
 
        void UpdateTrackSubscription(std::shared_ptr<SubscribeTrackHandler> track_handler);
 
        void PublishTrack(std::shared_ptr<PublishTrackHandler> track_handler);
 
        void UnpublishTrack(const std::shared_ptr<PublishTrackHandler>& track_handler);
 
        void PublishNamespace(std::shared_ptr<PublishNamespaceHandler> ns_handler, bool passive = false);
 
        void PublishNamespaceDone(const std::shared_ptr<PublishNamespaceHandler>& track_handler);
 
        void SubscribeNamespace(std::shared_ptr<SubscribeNamespaceHandler> handler);
 
        void UnsubscribeNamespace(const std::shared_ptr<SubscribeNamespaceHandler>& handler);
 
        void SendNamespace(const TrackNamespace& track_namespace);
 
        void SendNamespaceDone(const TrackNamespace& track_namespace);
 
        void FetchTrack(std::shared_ptr<FetchTrackHandler> track_handler);
 
        void CancelFetchTrack(std::shared_ptr<FetchTrackHandler> track_handler);
 
        void RequestTrackStatus(std::shared_ptr<TrackStatusHandler> handler);
 
        void GoAway(const std::string& new_session_uri,
                    std::chrono::milliseconds timeout,
                    std::optional<std::uint64_t> first_unprocessed_request_id = std::nullopt);
 
        void GoAway(const std::string& new_session_uri,
                    std::chrono::milliseconds timeout,
                    const std::shared_ptr<TrackHandler>& handler);
 
        // --BEGIN SERVER RELAY METHODS ----------------------------------------------------------------------
 
        void BindPublisherTrack(std::uint64_t src_id,
                                uint64_t request_id,
                                const std::shared_ptr<PublishTrackHandler>& track_handler,
                                bool ephemeral = false);
 
        void UnbindPublisherTrack(std::uint64_t src_id,
                                  const std::shared_ptr<PublishTrackHandler>& track_handler,
                                  bool send_publish_done = false);
 
        void BindFetchTrack(std::shared_ptr<PublishFetchHandler> track_handler);
 
        void UnbindFetchTrack(const std::shared_ptr<PublishFetchHandler>& track_handler);
 
        // --END SERVER RELAY METHODS ------------------------------------------------------------------------
 
      protected:
        Session() = delete;
 
        Session(const ClientConfig& cfg,
                std::shared_ptr<Transport> transport,
                std::shared_ptr<Connection> connection,
                std::shared_ptr<ClientCallbacks> callbacks,
                std::shared_ptr<Logger> logger);
 
        Session(const ServerConfig& cfg,
                std::shared_ptr<Transport> transport,
                std::shared_ptr<Connection> connection,
                std::shared_ptr<ServerCallbacks> callbacks,
                std::shared_ptr<Logger> logger);
 
        Session(const ClientConfig& cfg,
                std::shared_ptr<Transport> transport,
                std::shared_ptr<Connection> connection,
                std::shared_ptr<ClientCallbacks> callbacks,
                std::shared_ptr<Clock> clock,
                std::shared_ptr<Logger> logger);
 
        Session(const ServerConfig& cfg,
                std::shared_ptr<Transport> transport,
                std::shared_ptr<Connection> connection,
                std::shared_ptr<ServerCallbacks> callbacks,
                std::shared_ptr<Clock> clock,
                std::shared_ptr<Logger> logger);
 
        void OnStreamClosed(const std::shared_ptr<Stream>& stream, StreamClosedFlag flag) override;
 
      private:
        /*===================================================================*/
        // Transport Delegate/callback functions
        /*===================================================================*/
 
        void OnConnectionStatus(const std::shared_ptr<Connection>& connection, Connection::Status status) override;
 
        bool OnRecvStream(const std::shared_ptr<Stream>& stream) override;
 
        void OnRecvDgram(const std::shared_ptr<Connection>& connection) override;
 
        void OnConnectionMetricsSampled(const std::shared_ptr<Connection>& connection,
                                        MetricsTimeStamp sample_time,
 
                                        const QuicConnectionMetrics& quic_connection_metrics) override;
 
        void OnStreamMetricsStampled(const std::shared_ptr<Connection>& connection,
                                     MetricsTimeStamp sample_time,
                                     std::uint64_t stream_id,
                                     const QuicStreamMetrics& quic_stream_metrics,
                                     bool is_final) override;
 
        /*===================================================================*/
        // Private methods
        /*===================================================================*/
 
        void Init();
 
        void CheckReady(const std::shared_ptr<Connection>& connection);
 
        std::shared_ptr<Connection> GetConnection(std::uint64_t connection_id) const;
        std::shared_ptr<Connection> GetConnectionForStream(const std::shared_ptr<Stream>& stream) const;
 
        std::shared_ptr<Session> GetSharedPtr();
 
        bool ProcessRequestMessage(const std::shared_ptr<Connection>& connection,
                                   const std::shared_ptr<Stream>& stream,
                                   messages::ControlMessageType msg_type,
                                   BytesSpan msg_bytes);
 
        bool ProcessCtrlMessage(const std::shared_ptr<Connection>& connection,
                                const std::shared_ptr<Stream>& stream,
                                messages::ControlMessageType msg_type,
                                BytesSpan msg_bytes);
 
        void SetStatus(Status status);
 
        template<typename... Fields>
        void SendCtrlMsg(const std::shared_ptr<Stream>& stream, messages::ControlMessageType msg_type, Fields&&... args)
        {
            SendCtrlMsg(stream, false, msg_type, std::forward<Fields>(args)...);
        }
 
        template<typename... Fields>
        void SendCtrlMsg(const std::shared_ptr<Stream>& stream,
                         bool close_stream,
                         messages::ControlMessageType type,
                         Fields&&... args)
        {
            messages::Message msg = messages::Message{}.PrependType(type).ReserveLength();
 
            (msg.Append(args), ...);
 
            if (stream == nullptr) {
                throw std::logic_error("Stream cannot be null");
            }
 
            const auto connection = GetConnectionForStream(stream);
            if (!connection) {
                throw std::invalid_argument("Stream has no connection");
            }
            auto result = quic_transport_->Enqueue(
              connection, stream, msg.ToBytes(), 0, 2000, { true, close_stream, false, false });
 
            if (!result) {
                // TODO: Propagate better to callers and handle more specifically. Keeping the throw for now.
                throw TransportException(TransportError::kInvalidStreamId);
            }
        }
 
        void SendSetup(const std::shared_ptr<Connection>& connection, const std::string& connect_uri);
 
        /*===================================================================*/
        // Requests
        /*===================================================================*/
 
        void SendTrackStatusOk(const std::shared_ptr<Stream>& stream, const TrackStatusResponse& response);
 
        void SendSubscribeNamespaceOk(const std::shared_ptr<Stream>& stream);
 
        void SendSubscribeTracksOk(const std::shared_ptr<Stream>& stream) { SendSubscribeNamespaceOk(stream); }
 
        void SendPublishNamespaceOk(const std::shared_ptr<Stream>& stream) { SendSubscribeNamespaceOk(stream); }
 
        void SendRequestUpdateOk(const std::shared_ptr<Stream>& stream,
                                 std::optional<std::uint64_t> expires,
                                 const std::optional<messages::Location>& largest_object);
 
        // Prefer the above typed overloads.
        void SendRequestOk(const std::shared_ptr<Stream>& stream,
                           const messages::Parameters& params,
                           const messages::TrackExtensions& track_properties = {},
                           bool close_stream = false);
 
        void SendRequestUpdate(const std::shared_ptr<Stream>& stream,
                               TrackHash th,
                               std::optional<std::uint64_t> end_group_id,
                               std::uint8_t priority,
                               bool forward);
 
        void SendRequestError(const std::shared_ptr<Stream>& stream,
                              std::uint64_t request_id,
                              ErrorCode error,
                              std::chrono::milliseconds retry_interval,
                              const std::string& reason,
                              bool close_stream = true,
                              std::optional<Redirect> redirect = std::nullopt);
 
        /*===================================================================*/
        // Publish Namespace
        /*===================================================================*/
 
        bool SendPublishNamespace(const std::shared_ptr<Stream>& stream,
                                  std::uint64_t request_id,
                                  const TrackNamespace& track_namespace);
 
        /*===================================================================*/
        // Subscribe Namespace
        /*===================================================================*/
 
        bool SendSubscribeNamespace(const std::shared_ptr<Stream>& stream,
                                    std::uint64_t request_id,
                                    const TrackNamespace& prefix,
                                    const messages::Filter& filter,
                                    messages::ControlMessageType type);
 
        void SendNamespaceMessage(messages::ControlMessageType type, const TrackNamespace& track_namespace);
 
        /*===================================================================*/
        // Subscribe
        /*===================================================================*/
 
        bool SendSubscribe(const std::shared_ptr<Stream>& stream,
                           std::uint64_t request_id,
                           const FullTrackName& tfn,
                           TrackHash th,
                           std::uint8_t priority,
                           std::optional<messages::GroupOrder> group_order,
                           const messages::Filter& filter,
                           std::optional<std::chrono::milliseconds> delivery_timeout,
                           std::chrono::milliseconds rendezvous_timeout);
 
        void SendSubscribeOk(const std::shared_ptr<Stream>& stream,
                             std::uint64_t request_id,
                             uint64_t track_alias,
                             const SubscribeResponse& response);
 
        /*===================================================================*/
        // Publish
        /*===================================================================*/
 
        bool SendPublish(const std::shared_ptr<Stream>& stream,
                         std::uint64_t request_id,
                         const PublishAttributes& publish);
 
        void SendPublishDone(const std::shared_ptr<Stream>& stream,
                             std::uint64_t request_id,
                             messages::PublishDoneStatusCode status,
                             const std::string& reason);
 
        void SendPublishOk(const std::shared_ptr<Stream>& stream, const PublishOkAttributes& attributes);
 
        std::shared_ptr<Stream> FindSubscribeNamespaceStream(const TrackNamespace& track_namespace) const;
 
        std::optional<std::uint64_t> FindRequestIdByStream(const Stream& stream) QUICR_EXCLUDES(state_mutex_);
 
        bool IsCurrentRequestStream(std::uint64_t request_id, const std::shared_ptr<Stream>& stream)
          QUICR_EXCLUDES(state_mutex_);
 
        /*===================================================================*/
        // Track Status
        /*===================================================================*/
 
        void SendTrackStatus(const std::shared_ptr<Stream>& stream, std::uint64_t request_id, const FullTrackName& tfn);
 
        /*===================================================================*/
        // Fetch
        /*===================================================================*/
 
        void ResolveFetch(uint64_t request_id,
                          std::optional<messages::GroupOrder> group_order,
                          const FetchResponse& response);
 
        bool SendFetch(const std::shared_ptr<Stream>& stream,
                       std::uint64_t request_id,
                       const FullTrackName& tfn,
                       std::uint8_t priority,
                       std::optional<messages::GroupOrder> group_order,
                       const messages::Location& start_location,
                       const messages::FetchEndLocation& end_location);
 
        void SendJoiningFetch(const std::shared_ptr<Stream>& stream,
                              std::uint64_t request_id,
                              std::uint8_t priority,
                              std::optional<messages::GroupOrder> group_order,
                              std::uint64_t joining_request_id,
                              std::uint64_t joining_start,
                              bool absolute);
 
        void SendFetchOk(const std::shared_ptr<Stream>& stream,
                         messages::GroupOrder publisher_default_group_order,
                         bool end_of_track,
                         messages::Location end_location);
 
        /*===================================================================*/
        // Other member functions
        /*===================================================================*/
 
        void RemoveSubscribeTrack(SubscribeTrackHandler& handler, bool remove_handler = true);
 
        void CloseRequestHandler(std::uint64_t request_id, const std::shared_ptr<Stream>& stream, StreamClosedFlag flag)
          QUICR_EXCLUDES(state_mutex_);
 
        std::shared_ptr<PublishTrackHandler> GetPubTrackHandler(TrackHash& th);
 
        void RemoveTracksForConnectionClose(const std::shared_ptr<Connection>& connection);
 
        [[nodiscard]] uint64_t GetNextRequestID();
 
        bool BindRecvStream(Stream& stream);
 
 
        bool RecvCtrlMessage(const std::shared_ptr<Connection>& connection,
                             const std::shared_ptr<Stream>& stream,
                             bool is_request_stream);
 
        bool RecvSubgroupObject(Stream& stream, SubscribeTrackHandler& handler);
 
        bool RecvFetchObject(Stream& stream, SubscribeTrackHandler& handler);
 
 
        bool ForwardSubgroupBytes(Stream& stream, ForwardingSubscribeTrackHandler& handler);
 
        bool OnRecvSubgroup(std::uint64_t track_alias, Stream& stream);
 
        bool OnRecvFetch(std::uint64_t request_id, Stream& stream);
 
        std::shared_ptr<Stream> CreateStream(const std::shared_ptr<Connection>& connection,
                                             std::uint64_t request_id,
                                             uint8_t priority);
 
        bool Enqueue(const std::shared_ptr<Connection>& connection,
                     const std::shared_ptr<Stream>& stream,
                     std::shared_ptr<const std::vector<uint8_t>> bytes,
                     const uint8_t priority,
                     const uint32_t ttl_ms,
                     const Transport::EnqueueFlags flags);
 
      private:
        void SetConnectionAddedCallback(
          std::function<void(const std::shared_ptr<Connection>&, const std::shared_ptr<Session>&)> callback)
        {
            connection_added_callback_ = std::move(callback);
        }
 
        enum class MigrationRequestType : std::uint8_t
        {
            kSubscribe,
            kPublish,
            kSubscribeNamespace,
            kPublishNamespace,
            kFetch,
        };
 
        struct PendingMigration
        {
            MigrationRequestType type;
            std::shared_ptr<Stream> request_stream;
        };
 
        struct Migration
        {
            Redirect redirect;
            std::shared_ptr<Connection> source_connection;
            std::shared_ptr<Connection> candidate_connection;
            std::map<std::uint64_t, PendingMigration> pending;
        };
 
        using MigrationKey = std::pair<std::uint64_t, std::optional<std::uint64_t>>;
        using MigrationMap = std::map<MigrationKey, Migration>;
 
        void Migrate(const std::shared_ptr<Connection>& source_connection, const Redirect& redirect);
 
        void StartMigrationRequests(const std::shared_ptr<Connection>& connection);
 
        void CompleteMigration(const std::shared_ptr<Connection>& connection,
                               std::uint64_t request_id,
                               MigrationRequestType type,
                               std::optional<std::uint64_t> received_track_alias = std::nullopt);
 
        bool FailMigration(const std::shared_ptr<Connection>& connection, std::uint64_t request_id);
 
        bool MatchesRedirect(const TrackHandler& handler, const Redirect& redirect) const;
 
        MigrationMap::iterator FindMigration(const Connection& candidate) QUICR_REQUIRES(state_mutex_);
 
        void ResetSubgroup(const std::shared_ptr<Stream>& stream);
 
      private:
        mutable std::mutex connections_mutex_;
        std::map<std::uint64_t, std::shared_ptr<Connection>> connections_ QUICR_GUARDED_BY(connections_mutex_);
        MigrationMap migrations_ QUICR_GUARDED_BY(state_mutex_);
        std::shared_ptr<Connection> default_connection_ QUICR_GUARDED_BY(state_mutex_);
        std::function<void(const std::shared_ptr<Connection>&, const std::shared_ptr<Session>&)>
          connection_added_callback_;
 
        const std::shared_ptr<Callbacks> callbacks_;
 
        std::atomic<uint64_t> next_request_id_;
 
        std::map<std::uint64_t, SubscribeContext> recv_req_id;
 
        struct StreamRequest
        {
            std::uint64_t request_id;
 
            bool is_request_stream;
        };
 
        using StreamKey = std::pair<std::uint64_t, std::uint64_t>;
        std::map<StreamKey, StreamRequest> request_by_stream QUICR_GUARDED_BY(state_mutex_);
 
        std::vector<std::uint64_t>
          recv_publish_namespaces; 
 
        std::vector<std::uint64_t> recv_subscribe_namespaces;
 
        std::map<std::uint64_t, std::shared_ptr<TrackHandler>> request_handlers; 
 
        std::map<std::pair<std::uint64_t, std::uint64_t>, std::shared_ptr<SubscribeTrackHandler>>
          sub_by_recv_track_alias;
 
        std::map<std::uint64_t, std::map<std::uint64_t, std::shared_ptr<PublishTrackHandler>>> pub_tracks_by_name;
 
        std::map<std::uint64_t, std::map<uint64_t, std::shared_ptr<PublishTrackHandler>>> pub_tracks_by_track_alias
          QUICR_GUARDED_BY(state_mutex_);
 
        std::map<std::uint64_t, std::shared_ptr<PublishTrackHandler>>
          pub_fetch_tracks_by_request_id; 
 
        mutable std::mutex state_mutex_;
 
        const bool client_mode_;
 
        const std::shared_ptr<Logger> logger_;
 
        bool stop_{ false };
 
        const ServerConfig server_config_;
 
        const ClientConfig client_config_;
 
        std::atomic<Status> status_{ Status::kConnecting };
 
        const std::shared_ptr<Clock> clock_;
 
        const std::shared_ptr<Transport> quic_transport_; // **MUST** be last for proper order of destruction
 
        friend class PublishTrackHandler;
        friend class PublishFetchHandler;
        friend class SubscribeTrackHandler;
        friend class SessionManager;
    };
 
} // namespace quicr