rtpsession.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_NETWORK
#include <cstdint>
#include <promeki/atomic.h>
#include <promeki/list.h>
#include <promeki/mutex.h>
#include <promeki/objectbase.h>
#include <promeki/error.h>
#include <promeki/buffer.h>
#include <promeki/clockdomain.h>
#include <promeki/duration.h>
#include <promeki/enums_rtp.h>
#include <promeki/packetscheduler.h>
#include <promeki/queue.h>
#include <promeki/rtcppacket.h>
#include <promeki/socketaddress.h>
#include <promeki/ntptime.h>
#include <promeki/rtppacket.h>
#include <promeki/rtppacketbatch.h>
#include <promeki/packettransport.h>
#include <promeki/string.h>
#include <promeki/timestamp.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class Thread;
class RtpSeqTracker;
class RtpSeqReorderBuffer;
 
class RtpSession : public ObjectBase {
                PROMEKI_OBJECT(RtpSession, ObjectBase)
        public:
                RtpSession(ObjectBase *parent = nullptr);
 
                ~RtpSession() override;
 
                Error start(const SocketAddress &localAddr);
 
                Error start(PacketTransport *transport);
 
                Error start(PacketTransport *primary, PacketTransport *secondary);
 
                void stop();
 
                bool isRunning() const { return _running; }
 
                void setRemote(const SocketAddress &dest) { _remote = dest; }
 
                const SocketAddress &remote() const { return _remote; }
 
                void setRemoteSecondary(const SocketAddress &dest) { _remoteSecondary = dest; }
 
                const SocketAddress &remoteSecondary() const { return _remoteSecondary; }
 
                bool hasSecondaryLeg() const { return _transportSecondary != nullptr; }
 
                Error sendPacket(const Buffer &payload, uint32_t timestamp, uint8_t payloadType, bool marker = false);
 
                Error sendPackets(RtpPacketBatch &batch);
 
                struct StreamReceiver {
                                RtpPacket::Queue *outQueue = nullptr;
 
                                RtpSeqTracker *seqTracker = nullptr;
 
                                RtpSeqReorderBuffer *reorderBuffer = nullptr;
 
                                uint32_t clockRateHz = 0;
 
                                uint8_t payloadType = 0;
                };
 
                Error startReceiving(List<StreamReceiver> receivers,
                                     const String &threadName = "rtp-rx");
 
                void stopReceiving();
 
                bool isReceiving() const { return _receiving.value(); }
 
                void setReceivePollIntervalMs(unsigned int timeoutMs) {
                        _receivePollMs = timeoutMs == 0 ? 200 : timeoutMs;
                }
 
                unsigned int receivePollIntervalMs() const { return _receivePollMs; }
 
                Error setPacingRate(uint64_t bytesPerSec);
 
                void setScheduler(PacketScheduler::UPtr scheduler);
 
                PacketScheduler *scheduler() const { return _scheduler.get(); }
 
                void setSchedulerSecondary(PacketScheduler::UPtr scheduler);
 
                PacketScheduler *schedulerSecondary() const { return _schedulerSecondary.get(); }
 
                Error configureScheduler(const PacketScheduler::Spec &spec);
 
                uint32_t ssrc() const { return _ssrc; }
 
                void setSsrc(uint32_t ssrc) { _ssrc = ssrc; }
 
                uint16_t sequenceNumber() const { return _sequenceNumber; }
 
                void setPayloadType(uint8_t pt) { _payloadType = pt; }
 
                uint8_t payloadType() const { return _payloadType; }
 
                void setClockRate(uint32_t hz) { _clockRate = hz; }
 
                uint32_t clockRate() const { return _clockRate; }
 
                void setRtpAnchor(NtpTime captureNtp, uint32_t rtpTs);
 
                void setRtpAnchor(const ClockDomain &domain, uint32_t rtpTs);
 
                NtpTime anchorNtp() const;
 
                uint32_t anchorRtpTs() const;
 
                void noteRtpEmission(uint32_t rtpTs);
 
                bool hasEmissionRecord() const;
 
                const String &cname() const { return _cname; }
 
                void setCname(const String &cname) { _cname = cname; }
 
                Error emitRtcpSr(uint32_t senderPacketCount, uint32_t senderOctetCount);
 
                Error emitRtcpRr(const RtcpPacket::ReportBlock &block);
 
                Error emitRtcpBye();
 
                struct ReceivedSr {
                        NtpTime   ntp;
                        uint32_t  rtpTs = 0;
                        TimeStamp arrivedAt;
                        bool      valid = false;
                };
 
                ReceivedSr receivedSr() const;
 
                uint32_t srObservedCount() const;
 
                TimeStamp firstSrAt() const;
 
                NtpTime currentSrNtp() const;
 
                PacketTransport *transport() const { return _transport; }
 
                PROMEKI_SIGNAL(packetReceived, Buffer, uint32_t, uint8_t, bool);
 
                PROMEKI_SIGNAL(ssrcCollision, uint32_t);
 
                PROMEKI_SIGNAL(ssrcChange, uint32_t, uint32_t, uint8_t);
 
                PROMEKI_SIGNAL(byeReceived, uint32_t);
 
        private:
                class ReceiveThread;
                friend class ReceiveThread;
 
                void handleRtcp(const uint8_t *data, size_t size);
 
                void fillHeader(RtpPacket &pkt, uint8_t pt, bool marker, uint32_t timestamp);
                void fillTransportHeader(RtpPacket &pkt);
                void generateSsrc();
 
                PacketTransport      *_transport = nullptr;
                PacketTransport::UPtr _ownedTransport;
                PacketScheduler::UPtr _scheduler;
 
                // ST 2022-7 dual-leg state.  When _transportSecondary is
                // non-null sendPackets stamps the RTP header once and
                // fans the resulting Datagram list out to both
                // transports — leg-specific destinations are taken from
                // _remote (primary) and _remoteSecondary; each scheduler
                // is bound to its own transport so back-pressure on
                // one leg cannot stall the other.  Receive side: a
                // second receive thread services the secondary transport
                // and joins the primary thread on shutdown.  Both
                // recv threads dispatch through _dispatchMutex so the
                // per-stream seq tracker / SSRC pin / reorder buffer
                // stay coherent.
                PacketTransport      *_transportSecondary = nullptr;
                PacketTransport::UPtr _ownedTransportSecondary;
                PacketScheduler::UPtr _schedulerSecondary;
                SocketAddress         _remoteSecondary;
 
                bool                  _running = false;
                SocketAddress         _remote;
                uint32_t              _ssrc = 0;
                uint16_t              _sequenceNumber = 0;
                uint8_t               _payloadType = 96;
                uint32_t              _clockRate = 90000;
 
                // RTCP SR / SDES state.  CNAME is the SDES item every
                // SR-bearing compound packet carries.  The
                // capture-anchor is established once at openStream
                // time and refined by the first arriving Frame's
                // @ref Frame::captureTime; thereafter, @c emitRtcpSr
                // derives the SR's NTP from
                // @c anchorNtp + (lastEmissionRtpTs - anchorRtpTs) /
                // clockRate without sampling the system clock.
                // @c _hasEmission gates SR emission on whether any
                // packet has actually gone out, so the scheduler
                // never sends an SR for a session that has yet to
                // produce wire activity.  Mutex-guarded because the
                // noter runs on the per-stream TX thread while the
                // scheduler runs on its own thread.
                mutable Mutex _rtcpMutex;
                NtpTime       _anchorNtp;
                uint32_t      _anchorRtpTs = 0;
                uint32_t      _lastEmissionRtpTs = 0;
                bool          _hasEmission = false;
                String        _cname;
                uint32_t      _srObservedCount = 0;
                TimeStamp     _firstSrAt;
 
                // Most-recently parsed inbound SR.  The receive thread
                // demuxes RTCP from RTP via the second byte of every
                // datagram (PT in [200..223] → RTCP) and walks each
                // RTCP compound for SRs.  Reader-side helpers like
                // @ref RtpStreamClock pick this up via
                // @ref receivedSr to map any future RTP-TS on this
                // session to a wallclock instant for cross-stream
                // alignment.
                ReceivedSr _lastReceivedSr;
 
                // Receive path.  @c _streamReceivers is populated
                // by @ref startReceiving and consumed by the recv
                // socket thread; per-stream depacketizer threads
                // pull from the post-reorder queues each entry
                // points at.
                using ReceiveThreadUPtr = UniquePtr<ReceiveThread>;
                ReceiveThreadUPtr    _receiveThread;
                ReceiveThreadUPtr    _receiveThreadSecondary; 
                Mutex                _dispatchMutex;          
                List<StreamReceiver> _streamReceivers;
                Atomic<bool>         _receiving;
                unsigned int         _receivePollMs = 200;
 
                // Per-stream-receiver SSRC pin state.  Sized to
                // match @c _streamReceivers.size() at
                // @c startReceiving time.  Index-parallel — entry
                // @c i tracks the SSRC pin for receivers[i].
                struct SsrcPinState {
                                uint32_t  expectedSsrc = 0;
                                bool      pinned = false;
                                uint32_t  mismatchCount = 0;
                                TimeStamp mismatchFirstTime;
                };
                List<SsrcPinState> _ssrcPinStates;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_NETWORK