rtpaggregatorthread.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_NETWORK
#include <cstdint>
#include <functional>
#include <promeki/function.h>
#include <promeki/atomic.h>
#include <promeki/audiobuffer.h>
#include <promeki/audiodesc.h>
#include <promeki/clockdomain.h>
#include <promeki/duration.h>
#include <promeki/frame.h>
#include <promeki/framenumber.h>
#include <promeki/framerate.h>
#include <promeki/namespace.h>
#include <promeki/queue.h>
#include <promeki/rtpstreamclock.h>
#include <promeki/rxpayloadbundle.h>
#include <promeki/string.h>
#include <promeki/thread.h>
#include <promeki/timestamp.h>
 
PROMEKI_NAMESPACE_BEGIN
 
struct RtpAggregatorVideoStream {
                Queue<RxVideoFrame> *payloadQueue = nullptr;
 
                const Atomic<int64_t> *lastPacketArrivalNs = nullptr;
};
 
struct RtpAggregatorAudioStream {
                Queue<RxAudioChunk> *payloadQueue = nullptr;
 
                bool active = false;
 
                const AudioDesc *readerAudioDesc = nullptr;
 
                const RtpStreamClock *streamClock = nullptr;
 
                const bool *hasSr = nullptr;
 
                ClockDomain clockDomain;
};
 
struct RtpAggregatorDataStream {
                Queue<RxDataMessage> *payloadQueue = nullptr;
 
                bool active = false;
 
                ClockDomain clockDomain;
};
 
struct RtpAggregatorAncStream {
                Queue<RxAncFrame> *payloadQueue = nullptr;
 
                bool active = false;
 
                ClockDomain clockDomain;
};
 
struct RtpAggregatorContext {
                FrameRate frameRate;
 
                bool videoWatchdogEnabled = false;
 
                RtpAggregatorVideoStream video;
                RtpAggregatorAudioStream audio;
                RtpAggregatorDataStream  data;
                RtpAggregatorAncStream   anc;
 
                Queue<Frame> *readerQueue = nullptr;
 
                Function<void(Frame)> pushFrame;
};
 
class RtpAggregatorThread : public Thread {
        public:
                enum class Mode {
                        Video,      
                        AudioOnly,  
                        DataOnly,   
                        AncOnly     
                };
 
                static constexpr int kStallNFrames = 4;
 
                static constexpr unsigned int kPopCapMs = 50;
 
                static constexpr unsigned int kPopFloorMs = 5;
 
                RtpAggregatorThread(RtpAggregatorContext ctx, Mode mode,
                                    const String &name = String("RtpAggregator"));
 
                ~RtpAggregatorThread() override;
 
                RtpAggregatorThread(const RtpAggregatorThread &) = delete;
                RtpAggregatorThread &operator=(const RtpAggregatorThread &) = delete;
 
                void requestStop();
 
                bool isStopRequested() const { return _stopRequested.value(); }
 
                Mode mode() const { return _mode; }
 
                void runOnce(unsigned int popMs = kPopCapMs);
 
                const AudioBuffer &audioFifo() const { return _audioFifo; }
 
                TimeStamp emittedFrameCursor() const { return _emittedFrameCursor; }
 
                bool inWatchdog() const { return _inWatchdog; }
 
        protected:
                void run() override;
 
        private:
                Duration frameDuration() const;
 
                void runVideoMode();
                void runAudioOnlyMode();
                void runDataOnlyMode();
                void runAncOnlyMode();
 
                void drainAudioIntoFifoBefore(const TimeStamp &windowEnd);
 
                bool drainDataBefore(const TimeStamp &windowEnd, RxDataMessage &out);
 
                bool drainAncBefore(const TimeStamp &windowEnd, RxAncFrame &out);
 
                bool stepVideoMode(unsigned int popMs);
 
                bool stepAudioOnlyMode(unsigned int popMs);
 
                bool stepDataOnlyMode(unsigned int popMs);
 
                bool stepAncOnlyMode(unsigned int popMs);
 
                void emitFrameForVideo(RxVideoFrame video, const Duration &fd);
                void emitWatchdogFrame(const Duration &fd);
                void emitAudioOnlyAtFrameRate(const RxAudioChunk &c);
                void emitAudioOnlyPerChunk(const RxAudioChunk &c);
 
                bool ensureAudioFifo(const AudioDesc &wireDesc);
 
                unsigned int videoPopMs(const Duration &fd) const;
 
                RtpAggregatorContext _ctx;
                Mode                 _mode;
                Atomic<bool>         _stopRequested;
 
                AudioBuffer _audioFifo;
 
                uint32_t _audioFifoFrontRtpTs = 0;
 
                bool _audioFifoHasFront = false;
 
                FrameNumber _videoFrameIndex{0};
 
                TimeStamp _emittedFrameCursor;
 
                bool _inWatchdog = false;
 
                RxDataMessage _pendingData;
 
                bool _hasPendingData = false;
 
                RxAncFrame _pendingAnc;
 
                bool _hasPendingAnc = false;
 
                bool _firstVideoSeen = false;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_NETWORK