rtpmediaio.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_NETWORK
#include <promeki/atomic.h>
#include <promeki/audiobuffer.h>
#include <promeki/audiodesc.h>
#include <promeki/clockdomain.h>
#include <promeki/dedicatedthreadmediaio.h>
#include <promeki/eui64.h>
#include <promeki/macaddress.h>
#include <promeki/frame.h>
#include <promeki/histogram.h>
#include <promeki/imagedesc.h>
#include <promeki/mediaiofactory.h>
#include <promeki/mutex.h>
#include <promeki/namespace.h>
#include <promeki/ntptime.h>
#include <promeki/pacinggate.h>
#include <promeki/pcmaudiopayload.h>
#include <promeki/pixelformat.h>
#include <promeki/queue.h>
#include <promeki/rtcpscheduler.h>
#include <promeki/rtpaggregatorthread.h>
#include <promeki/rtpancdepacketizerthread.h>
#include <promeki/rtpaudiodepacketizerthread.h>
#include <promeki/rtpaudiopacketizerthread.h>
#include <promeki/rtpaudiotxthread.h>
#include <promeki/rtpdatadepacketizerthread.h>
#include <promeki/rtpdepacketizerthread.h>
#include <promeki/rtpvideodepacketizerthread.h>
#include <promeki/rtppacket.h>
#include <promeki/rtppacketizerthread.h>
#include <promeki/rtpseqreorderbuffer.h>
#include <promeki/rtpseqtracker.h>
#include <promeki/rtpstreamclock.h>
#include <promeki/rtptxthread.h>
#include <promeki/rxpayloadbundle.h>
#include <promeki/sdpsession.h>
#include <promeki/socketaddress.h>
#include <promeki/string.h>
#include <promeki/timestamp.h>
#include <promeki/videopayload.h>
#include <promeki/waitcondition.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class RtpSession;
class RtpPayload;
class UdpSocketTransport;
class Thread;
 
class RtpMediaIO : public DedicatedThreadMediaIO {
                PROMEKI_OBJECT(RtpMediaIO, DedicatedThreadMediaIO)
        public:
                static inline const MediaIOStats::ID StatsFramesSent{"FramesSent"};
                static inline const MediaIOStats::ID StatsPacketsSent{"PacketsSent"};
                static inline const MediaIOStats::ID StatsBytesSent{"BytesSent"};
                static inline const MediaIOStats::ID StatsFramesReceived{"FramesReceived"};
                static inline const MediaIOStats::ID StatsPacketsReceived{"PacketsReceived"};
                static inline const MediaIOStats::ID StatsBytesReceived{"BytesReceived"};
                static inline const MediaIOStats::ID StatsTxVideoFrameIntervalUs{"TxVideoFrameIntervalUs"};
                static inline const MediaIOStats::ID StatsTxVideoSendDurationUs{"TxVideoSendDurationUs"};
                static inline const MediaIOStats::ID StatsRxVideoPacketIntervalUs{"RxVideoPacketIntervalUs"};
                static inline const MediaIOStats::ID StatsRxVideoFrameIntervalUs{"RxVideoFrameIntervalUs"};
                static inline const MediaIOStats::ID StatsRxVideoFrameAssembleUs{"RxVideoFrameAssembleUs"};
                static inline const MediaIOStats::ID StatsAudioSilencePacketsEmitted{"AudioSilencePacketsEmitted"};
                static inline const MediaIOStats::ID StatsAudioSilenceSamplesEmitted{"AudioSilenceSamplesEmitted"};
 
                // ----------------------------------------------------------
                // Reader-side per-stream RFC 3550 §A counters,
                // aggregated across every active reader stream
                // (video + audio + data).  All published by
                // @c executeCmd(MediaIOCommandStats); see
                // @c devplan/network/rtp-rx.md for the full
                // ReaderStream::Stats block these correspond to.
                // ----------------------------------------------------------
 
                static inline const MediaIOStats::ID StatsRxExtendedHighestSeq{"RxExtendedHighestSeq"};
                static inline const MediaIOStats::ID StatsRxPacketsExpected{"RxPacketsExpected"};
                static inline const MediaIOStats::ID StatsRxCumulativeLost{"RxCumulativeLost"};
                static inline const MediaIOStats::ID StatsRxFractionLost{"RxFractionLost"};
                static inline const MediaIOStats::ID StatsRxDuplicatePackets{"RxDuplicatePackets"};
                static inline const MediaIOStats::ID StatsRxReorderedPackets{"RxReorderedPackets"};
                static inline const MediaIOStats::ID StatsRxInterarrivalJitter{"RxInterarrivalJitter"};
 
                static inline const MediaIOStats::ID StatsRxSsrcChanges{"RxSsrcChanges"};
 
                static inline const MediaIOStats::ID StatsRxReorderEmittedInOrder{"RxReorderEmittedInOrder"};
                static inline const MediaIOStats::ID StatsRxReorderEmittedOnDeadline{"RxReorderEmittedOnDeadline"};
                static inline const MediaIOStats::ID StatsRxReorderDroppedOverflow{"RxReorderDroppedOverflow"};
                static inline const MediaIOStats::ID StatsRxReorderDroppedDuplicate{"RxReorderDroppedDuplicate"};
 
                static inline const MediaIOStats::ID StatsRxVideoQueueDepth{"RxVideoQueueDepth"};
                static inline const MediaIOStats::ID StatsRxAudioQueueDepth{"RxAudioQueueDepth"};
                static inline const MediaIOStats::ID StatsRxDataQueueDepth{"RxDataQueueDepth"};
                static inline const MediaIOStats::ID StatsRxReaderQueueDepth{"RxReaderQueueDepth"};
 
                static inline const MediaIOStats::ID StatsRxFramesReassembled{"RxFramesReassembled"};
                static inline const MediaIOStats::ID StatsRxFramesDroppedValidate{"RxFramesDroppedValidate"};
                static inline const MediaIOStats::ID StatsRxFramesWaitingParamSets{"RxFramesWaitingParamSets"};
                static inline const MediaIOStats::ID StatsRxFramesDroppedSsrcReset{"RxFramesDroppedSsrcReset"};
 
                static inline const MediaIOStats::ID StatsRxSrObserved{"RxSrObserved"};
                static inline const MediaIOStats::ID StatsRxLastSrAgeUs{"RxLastSrAgeUs"};
                static inline const MediaIOStats::ID StatsRxFirstSrLatencyUs{"RxFirstSrLatencyUs"};
 
                static inline const MediaIOParamsID ParamGetSdp{"GetSdp"};
                static inline const MediaIOParamsID ParamSdp{"Sdp"};
 
                RtpMediaIO(ObjectBase *parent = nullptr);
 
                ~RtpMediaIO() override;
 
                uint64_t objectId() const { return _objectId; }
 
                static String buildDefaultCname(int64_t pid, uint64_t objectId, const String &host);
 
                static String pickEgressHostForCname(const SocketAddress &destination);
 
        protected:
                Error executeCmd(MediaIOCommandOpen &cmd) override;
                Error executeCmd(MediaIOCommandClose &cmd) override;
                Error executeCmd(MediaIOCommandRead &cmd) override;
                Error executeCmd(MediaIOCommandWrite &cmd) override;
                Error executeCmd(MediaIOCommandParams &cmd) override;
                Error executeCmd(MediaIOCommandStats &cmd) override;
                Error executeCmd(MediaIOCommandSetClock &cmd) override;
 
                // Wakes the reader-side executeCmd(Read) loop so close()
                // can drain a strand parked on _readerQueue.pop().
                void cancelBlockingWork() override;
 
        private:
 
                // describe() / proposeInput overrides intentionally
                // omitted for v1: each RFC 9134 / RFC 2435 / RFC 4175
                // payload type has its own accepted-shape constraints
                // (e.g. JPEG payload = 8-bit YUV422 only; raw payload
                // = configurable subsampling / bit depth; L16 audio =
                // 16-bit BE PCM at fixed channel counts).  Encoding
                // those rules per-payload is its own follow-up — for
                // now the planner's open() fallback inspects the SDP
                // (when configured) to learn the live shape, and
                // bridges insert as needed via the configured payload
                // type's own runtime checks.
 
                // Per-stream packetizer + TX worker classes.  Each
                // active stream gets one of each: the packetizer
                // pulls Frames off the strand-side @c PayloadQueue
                // and emits payload-bytes-only packets onto the
                // sink-side queue; the TX thread pops the latter,
                // stamps the full RTP header, and dispatches to
                // the wire.  Packetizer and TX talk only through
                // their own per-stream @c Queue, so stream-level
                // jitter (heavy IDR encode, audio cadence) never
                // bleeds across stream boundaries.  Concrete
                // subclasses live as nested classes inside
                // @c rtpmediaio.cpp because they need access to
                // RtpMediaIO state (parameter-set cache, pacing
                // mode, SDP sprop refresh, etc.).
                class VideoPacketizerThread;
                class AudioPacketizerThread;
                class DataPacketizerThread;
                class VideoTxThread;
                class AudioTxThread;
                class DataTxThread;
 
                struct Stream {
                                Stream() = default;
                                Stream(Stream &&o) noexcept
                                    : transport(o.transport),
                                      session(o.session),
                                      payload(o.payload),
                                      destination(std::move(o.destination)),
                                      payloadType(o.payloadType),
                                      clockRate(o.clockRate),
                                      dscp(o.dscp),
                                      ssrc(o.ssrc),
                                      mediaType(std::move(o.mediaType)),
                                      rtpmap(std::move(o.rtpmap)),
                                      fmtp(std::move(o.fmtp)),
                                      active(o.active),
                                      clockDomain(o.clockDomain),
                                      tsRefClkMode(o.tsRefClkMode),
                                      ptpGrandmaster(o.ptpGrandmaster),
                                      ptpDomain(o.ptpDomain),
                                      refClockLocalMac(std::move(o.refClockLocalMac)),
                                      mediaClkOffset(o.mediaClkOffset) {
                                        // Null pointers on the moved-
                                        // from instance so a stray
                                        // reset can't double-delete.
                                        o.transport = nullptr;
                                        o.session = nullptr;
                                        o.payload = nullptr;
                                }
                                Stream(const Stream &) = delete;
                                Stream &operator=(const Stream &) = delete;
                                Stream &operator=(Stream &&) = delete;
                                UdpSocketTransport *transport = nullptr;
                                RtpSession         *session = nullptr;
                                RtpPayload         *payload = nullptr;
                                SocketAddress       destination;
                                uint8_t             payloadType = 0;
                                uint32_t            clockRate = 90000;
                                int                 dscp = 0;
                                uint32_t            ssrc = 0;
                                String              mediaType; 
                                String              rtpmap;    
                                String              fmtp;      
                                bool                active = false;
                                ClockDomain         clockDomain;    
                                RtpRefClockMode     tsRefClkMode = RtpRefClockMode::None; 
                                EUI64               ptpGrandmaster; 
                                uint8_t             ptpDomain = 0;  
                                MacAddress          refClockLocalMac; 
                                int32_t             mediaClkOffset = 0; 
                };
 
                struct WriterStream : Stream {
                                WriterStream() = default;
                                WriterStream(WriterStream &&o) noexcept
                                    : Stream(std::move(o)),
                                      packetizer(o.packetizer),
                                      tx(o.tx),
                                      packetsSent(o.packetsSent.value()),
                                      bytesSent(o.bytesSent.value()),
                                      senderOctets(o.senderOctets.value()),
                                      txFrameInterval(std::move(o.txFrameInterval)),
                                      txSendDuration(std::move(o.txSendDuration)),
                                      txLastSendStart(o.txLastSendStart),
                                      txHasLastSend(o.txHasLastSend) {
                                        o.packetizer = nullptr;
                                        o.tx = nullptr;
                                }
                                WriterStream(const WriterStream &) = delete;
                                WriterStream &operator=(const WriterStream &) = delete;
                                WriterStream &operator=(WriterStream &&) = delete;
                                RtpPacketizerThread *packetizer = nullptr;
                                RtpTxThread         *tx = nullptr;
                                Atomic<int64_t>     packetsSent{0};
                                Atomic<int64_t>     bytesSent{0};
                                Atomic<int64_t>     senderOctets{0}; 
                                Histogram           txFrameInterval; 
                                Histogram           txSendDuration;  
                                TimeStamp           txLastSendStart; 
                                bool                txHasLastSend = false;
                };
 
                struct ReaderStream : Stream {
                                ReaderStream() = default;
                                ReaderStream(ReaderStream &&o) noexcept
                                    : Stream(std::move(o)),
                                      packetsReceived(o.packetsReceived.value()),
                                      bytesReceived(o.bytesReceived.value()),
                                      framesReceived(o.framesReceived),
                                      packetsLost(o.packetsLost.value()),
                                      readerImageDesc(std::move(o.readerImageDesc)),
                                      readerAudioDesc(std::move(o.readerAudioDesc)),
                                      reasmTimestamp(o.reasmTimestamp),
                                      reasmHasTimestamp(o.reasmHasTimestamp),
                                      reasmLastSeq(o.reasmLastSeq),
                                      reasmHaveLastSeq(o.reasmHaveLastSeq),
                                      reasmPackets(std::move(o.reasmPackets)),
                                      rxPacketInterval(std::move(o.rxPacketInterval)),
                                      rxFrameInterval(std::move(o.rxFrameInterval)),
                                      rxFrameAssembleTime(std::move(o.rxFrameAssembleTime)),
                                      rxLastPacketTime(o.rxLastPacketTime),
                                      rxLastFrameTime(o.rxLastFrameTime),
                                      rxFrameStartTime(o.rxFrameStartTime),
                                      rxHasLastPacket(o.rxHasLastPacket),
                                      rxHasLastFrame(o.rxHasLastFrame),
                                      rxHasFrameStart(o.rxHasFrameStart),
                                      streamClock(std::move(o.streamClock)),
                                      lastSrArrivedAt(o.lastSrArrivedAt),
                                      hasSr(o.hasSr),
                                      seqTracker(std::move(o.seqTracker)),
                                      reorderBuffer(std::move(o.reorderBuffer)),
                                      reorderQueue(std::move(o.reorderQueue)),
                                      depacketizer(std::move(o.depacketizer)),
                                      ssrcChanges(o.ssrcChanges.value()),
                                      framesReassembled(o.framesReassembled.value()),
                                      framesDroppedValidate(o.framesDroppedValidate.value()),
                                      framesWaitingParamSets(o.framesWaitingParamSets.value()),
                                      framesDroppedSsrcReset(o.framesDroppedSsrcReset.value()) {}
                                ReaderStream(const ReaderStream &) = delete;
                                ReaderStream &operator=(const ReaderStream &) = delete;
                                ReaderStream &operator=(ReaderStream &&) = delete;
                                Atomic<int64_t> packetsReceived{0};
                                Atomic<int64_t> bytesReceived{0};
                                FrameCount      framesReceived{0};
                                Atomic<int64_t> packetsLost{0};
                                ImageDesc       readerImageDesc;
                                AudioDesc       readerAudioDesc;
                                uint32_t        reasmTimestamp = 0;
                                bool            reasmHasTimestamp = false;
                                uint16_t        reasmLastSeq = 0;
                                bool            reasmHaveLastSeq = false;
                                RtpPacket::List reasmPackets;
                                Histogram       rxPacketInterval;    
                                Histogram       rxFrameInterval;     
                                Histogram       rxFrameAssembleTime; 
                                TimeStamp       rxLastPacketTime;    
                                TimeStamp       rxLastFrameTime;     
                                TimeStamp       rxFrameStartTime;    
                                bool            rxHasLastPacket = false;
                                bool            rxHasLastFrame = false;
                                bool            rxHasFrameStart = false;
                                RtpStreamClock streamClock;
                                TimeStamp lastSrArrivedAt;
                                bool hasSr = false;
 
                                UniquePtr<RtpSeqTracker> seqTracker;
 
                                UniquePtr<RtpSeqReorderBuffer> reorderBuffer;
 
                                UniquePtr<RtpPacket::Queue> reorderQueue;
 
                                UniquePtr<RtpDepacketizerThread> depacketizer;
 
                                Atomic<uint32_t> resetEpoch{0};
 
                                Atomic<int64_t> lastPacketArrivalNs{0};
 
                                bool wireSilenceEosSignaled = false;
 
                                Atomic<int64_t> ssrcChanges{0};
 
                                Atomic<int64_t> framesReassembled{0};
 
                                Atomic<int64_t> framesDroppedValidate{0};
 
                                Atomic<int64_t> framesWaitingParamSets{0};
 
                                Atomic<int64_t> framesDroppedSsrcReset{0};
                };
 
                struct VideoStream : WriterStream {
                                VideoStream() = default;
                                VideoStream(VideoStream &&o) noexcept
                                    : WriterStream(std::move(o)),
                                      imageDesc(std::move(o.imageDesc)),
                                      cachedSps(std::move(o.cachedSps)),
                                      cachedPps(std::move(o.cachedPps)),
                                      cachedVps(std::move(o.cachedVps)) {}
                                VideoStream(const VideoStream &) = delete;
                                VideoStream &operator=(const VideoStream &) = delete;
                                VideoStream &operator=(VideoStream &&) = delete;
                                ImageDesc imageDesc;
                                Buffer cachedSps;
                                Buffer cachedPps;
                                Buffer cachedVps;
                };
 
                struct DataStream : WriterStream {
                                DataStream() = default;
                                DataStream(DataStream &&o) noexcept = default;
                                DataStream(const DataStream &) = delete;
                                DataStream &operator=(const DataStream &) = delete;
                                DataStream &operator=(DataStream &&) = delete;
                };
 
                struct AudioStream : WriterStream {
                                AudioStream() = default;
                                AudioStream(AudioStream &&o) noexcept
                                    : WriterStream(std::move(o)),
                                      storageDesc(std::move(o.storageDesc)),
                                      packetSamples(o.packetSamples),
                                      packetBytes(o.packetBytes),
                                      packetTimeUs(o.packetTimeUs),
                                      prerollSamples(o.prerollSamples),
                                      silencePacketsEmitted(
                                              o.silencePacketsEmitted.value()),
                                      silenceSamplesEmitted(
                                              o.silenceSamplesEmitted.value()) {}
                                AudioStream(const AudioStream &) = delete;
                                AudioStream &operator=(const AudioStream &) = delete;
                                AudioStream &operator=(AudioStream &&) = delete;
                                AudioDesc storageDesc;
                                size_t packetSamples = 0;
                                size_t packetBytes = 0;
                                int packetTimeUs = 0;
                                size_t prerollSamples = 0;
                                Atomic<int64_t> silencePacketsEmitted{0};
                                Atomic<int64_t> silenceSamplesEmitted{0};
                };
 
                struct VideoReaderStream : ReaderStream {
                                VideoReaderStream() = default;
                                VideoReaderStream(VideoReaderStream &&o) noexcept
                                    : ReaderStream(std::move(o)),
                                      payloadQueue(std::move(o.payloadQueue)) {}
                                VideoReaderStream(const VideoReaderStream &) = delete;
                                VideoReaderStream &operator=(const VideoReaderStream &) = delete;
                                VideoReaderStream &operator=(VideoReaderStream &&) = delete;
                                UniquePtr<Queue<RxVideoFrame>> payloadQueue;
                };
 
                struct AudioReaderStream : ReaderStream {
                                AudioReaderStream() = default;
                                AudioReaderStream(AudioReaderStream &&o) noexcept
                                    : ReaderStream(std::move(o)),
                                      fifo(std::move(o.fifo)),
                                      payloadQueue(std::move(o.payloadQueue)) {}
                                AudioReaderStream(const AudioReaderStream &) = delete;
                                AudioReaderStream &operator=(const AudioReaderStream &) = delete;
                                AudioReaderStream &operator=(AudioReaderStream &&) = delete;
                                AudioBuffer fifo;
                                UniquePtr<Queue<RxAudioChunk>> payloadQueue;
                };
 
                struct DataReaderStream : ReaderStream {
                                DataReaderStream() = default;
                                DataReaderStream(DataReaderStream &&o) noexcept
                                    : ReaderStream(std::move(o)),
                                      payloadQueue(std::move(o.payloadQueue)),
                                      ancPayloadQueue(std::move(o.ancPayloadQueue)) {}
                                DataReaderStream(const DataReaderStream &) = delete;
                                DataReaderStream &operator=(const DataReaderStream &) = delete;
                                DataReaderStream &operator=(DataReaderStream &&) = delete;
                                UniquePtr<Queue<RxDataMessage>> payloadQueue;
                                UniquePtr<Queue<RxAncFrame>> ancPayloadQueue;
                };
 
                static constexpr size_t VideoPayloadQueueDepth = 4;
                static constexpr size_t AudioPayloadQueueDepth = 100;
                static constexpr size_t DataPayloadQueueDepth = 8;
 
                Error configureVideoStream(const MediaIO::Config &cfg, const MediaDesc &mediaDesc);
                Error configureAudioStream(const MediaIO::Config &cfg, const MediaDesc &mediaDesc);
                Error configureDataStream(const MediaIO::Config &cfg);
 
                void applyClockReferenceConfig(const MediaIO::Config &cfg);
 
                Error openStream(WriterStream &s, bool enableMulticastLoopback);
                Error openReaderStream(ReaderStream &s, bool enableMulticastLoopback);
 
                Error injectParameterSets(const uint8_t *data, size_t size, Buffer &healed);
 
                bool paceVideoFrame();
 
                // sendVideo / sendAudio / sendData were per-stream
                // strand-side helpers in the previous architecture.
                // They are gone in Phase 2; their packetization
                // logic lives in the per-stream
                // @c VideoPacketizerThread / @c AudioPacketizerThread
                // / @c DataPacketizerThread (declared as nested
                // classes inside @c rtpmediaio.cpp), and their wire-
                // pacing logic lives in the matching
                // @c VideoTxThread / @c AudioTxThread /
                // @c DataTxThread.
 
                void refreshStreamClock(ReaderStream &s);
 
                TimeStamp ntpToSteady(const NtpTime &ntp) const;
 
                // Reader path.
                Error applySdp(const SdpSession &sdp, MediaIO::Config &cfg, MediaDesc &mediaDesc);
                Error openAllReaders();
                void  pushReaderFrame(Frame frame);
 
                void  buildSdp();
                Error writeSdpFile(const String &path);
 
                void resetStreamCommon(Stream &s);
                void resetWriterStream(WriterStream &s);
                void resetReaderStream(ReaderStream &s);
                void resetAll();
 
                RtcpSchedulerContext buildRtcpSchedulerContext();
 
                // Per-mode stream lists — writer-mode populates the
                // first three, reader-mode the last three.  Reader
                // and writer modes are mutually exclusive on a
                // single @c RtpMediaIO instance, so only one set is
                // populated at any given time.  Each kind is a list
                // (instead of a single slot) so the routing, SDP
                // builder, RTCP scheduler, and per-stream threads
                // can iterate uniformly regardless of how many
                // streams of each kind a session carries.  Today's
                // @c configureVideoStream / @c configureAudioStream /
                // @c configureDataStream populate one entry per kind
                // on the matching list (per @c _readerMode); the
                // collective shape is ready for multi-stream config
                // that lands later.
                List<VideoStream>       _videos;
                List<AudioStream>       _audios;
                List<DataStream>        _datas;
                List<VideoReaderStream> _videoReaders;
                List<AudioReaderStream> _audioReaders;
                List<DataReaderStream>  _dataReaders;
 
                // Transport-global config
                SocketAddress _localAddress;
                String        _sessionName;
                String        _sessionOrigin;
                String        _multicastInterface;
                int           _multicastTTL = 0;
                int           _recvBufferBytes = 0;
                int           _sendBufferBytes = 0;
                Enum          _pacingMode;
                Enum          _dataFormat;
 
                // Runtime
                FrameRate  _frameRate;
                FrameCount _frameCount{0};
                FrameCount _framesSent{0};
 
                // Mode
                bool _readerMode = false;
 
                // RFC 4175 wire-format PixelFormat.  When the input
                // pixel format doesn't match what RFC 4175 expects
                // on the wire (e.g. YUYV vs UYVY), the
                // VideoPacketizerThread calls
                // UncompressedVideoPayload::convert() to the wire
                // format before packing.  Invalid means no conversion
                // needed.
                PixelFormat _videoWirePixelFormat;
 
                // Reader runtime
                Queue<Frame> _readerQueue;
                int               _readerMaxDepth = 4;
                int               _readerJitterMs = 50;
                int               _wireSilenceTimeoutMs = 0; 
                bool              _videoWatchdogEnabled = false; 
                FrameCount        _readerFramesReceived{0};
                // Set by cancelBlockingWork() so the executeCmd(Read)
                // pop loop can break out of its short-timeout polling
                // when MediaIO::close is unwinding the strand.  Cleared
                // at every Open so a closed-then-reopened RtpMediaIO
                // doesn't carry the previous instance's cancellation
                // forward.
                Atomic<bool>      _readCancelled{false};
 
                TimeStamp _readerSteadyAnchor;
                NtpTime   _readerNtpAnchor;
                bool      _readerHasAnchor = false;
                TimeStamp _openedAt;
                UniquePtr<RtpAggregatorThread> _aggregator;
 
                // SDP — the active session description built at
                // open time.  Reader mode leaves this empty (the
                // reader consumes an externally-supplied SDP via
                // RtpSdp); writer mode populates it so the
                // GetSdp params command and the RtpSaveSdpPath
                // export path can serve it.
                SdpSession _sdpSession;
                String     _sdpPath;
 
                // RTCP — one scheduler thread per RtpMediaIO.  Wakes
                // every @c _rtcpIntervalMs and emits an SR + SDES
                // compound on every active writer stream's
                // RtpSession plus an RR for every active reader
                // stream.  The wallclock anchor captured at open
                // time is shared across all streams so a single
                // receiver-side observation of any stream's first
                // SR is sufficient for cross-stream correlation.
                // Disabled in reader mode (we are not a sender
                // there) and when @c MediaConfig::RtpRtcpEnabled
                // is @c false.  The scheduler itself lives in
                // @c rtcpscheduler.h; this io populates an
                // @c RtcpSchedulerContext at @c executeCmd(Open)
                // time and hands it to the scheduler's constructor.
                UniquePtr<RtcpScheduler> _rtcpScheduler;
                bool           _rtcpEnabled = true;
                int            _rtcpIntervalMs = 5000;
                String         _rtcpCname;
 
                // Process-local monotonic counter that gives every
                // RtpMediaIO a distinct id within one process — see
                // objectId() and buildDefaultCname().  Starts at 0,
                // first instance gets 1.
                static Atomic<uint64_t> _nextObjectId;
                uint64_t                _objectId = 0;
 
                // SR-anchor seeding gate.  At @c openStream time
                // every active session is anchored with
                // @c (NtpTime::now(), 0) so an SR can be emitted
                // even before the first frame arrives (better than
                // a structurally-invalid SR for late receivers).
                // The very first @c executeCmd(Write) refines the
                // anchor from the Frame's @ref Frame::captureTime —
                // we use a compare-exchange on this flag so even
                // future changes that take the Write path off the
                // single-threaded strand still seed exactly once
                // per opening.
                Atomic<bool> _anchorSeeded;
 
                // External writer-mode video pacing — null clock means
                // the upstream pump's natural cadence is the only
                // timing source.  Set via
                // executeCmd(MediaIOCommandSetClock); read on the
                // dedicated worker thread in executeCmd(Write) (same
                // thread as the setter, no synchronization required).
                // Audio is not paced separately by the gate — AES67
                // packet timing is governed by the per-packet RTP
                // timestamp stride that the audio FIFO maintains.
                PacingGate _videoGate;
 
                // Out-of-band parameter sets for H.264 / HEVC writers,
                // populated lazily by injectParameterSets() the first
                // time a complete set passes through the bitstream
                // (typically frame 0).  Stored as the base64-encoded
                // SDP-ready string — for H.264 it's
                // @c "<sps-base64>,<pps-base64>" (matches RFC 6184
                // @c sprop-parameter-sets); for HEVC it's the three
                // separate sprop-vps / -sps / -pps strings (the bool
                // tracks whether they have been populated).  Empty
                // until the encoder's first IDR / IRAP flows through;
                // once populated, buildSdp() embeds them in the
                // @c a=fmtp line so a receiver that reads the SDP
                // (e.g. ffplay) can populate its decoder's extradata
                // before the first packet arrives.  ffmpeg's H.264 RTP
                // demuxer requires this — without it, a receiver that
                // joins after the first IDR fails its initial codec
                // probe and never recovers, even with in-band
                // parameter sets repeated on every IDR.
                String _h264SpropParameterSets;
                String _h265SpropVps;
                String _h265SpropSps;
                String _h265SpropPps;
 
                void refreshSdpSprop();
};
 
class RtpFactory : public MediaIOFactory {
        public:
                RtpFactory() = default;
 
                String name() const override { return String("Rtp"); }
                String displayName() const override { return String("RTP Stream"); }
                String description() const override {
                        return String("RTP video + audio + metadata reader / writer "
                                      "(MJPEG / JPEG XS / H.264 / H.265 / raw / L16 / JSON)");
                }
                // An SDP file on disk implies the Rtp reader; writers
                // never open via a filesystem path, but extension-based
                // dispatch still uses this list so `-i foo.sdp` in
                // mediaplay picks the Rtp backend automatically.
                StringList extensions() const override {
                        return {String("sdp")};
                }
 
                bool canBeSource() const override { return true; }
                bool canBeSink() const override { return true; }
 
                bool            canHandleDevice(IODevice *device) const override;
                Config::SpecMap configSpecs() const override;
 
                MediaIO *create(const Config &config, ObjectBase *parent = nullptr) const override;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_NETWORK