rtcpscheduler.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/list.h>
#include <promeki/mutex.h>
#include <promeki/namespace.h>
#include <promeki/rtcppacket.h>
#include <promeki/rtpseqtracker.h>
#include <promeki/rtpsession.h>
#include <promeki/string.h>
#include <promeki/thread.h>
#include <promeki/waitcondition.h>
 
PROMEKI_NAMESPACE_BEGIN
 
struct RtcpSchedulerWriterStream {
                bool active = false;
 
                String mediaType;
 
                RtpSession *session = nullptr;
 
                const Atomic<int64_t> *packetsSent = nullptr;
 
                const Atomic<int64_t> *senderOctets = nullptr;
};
 
struct RtcpSchedulerReaderStream {
                bool active = false;
 
                String mediaType;
 
                RtpSession *session = nullptr;
 
                const RtpSeqTracker *seqTracker = nullptr;
 
                const Atomic<int64_t> *lastPacketArrivalNs = nullptr;
 
                bool *wireSilenceEosSignaled = nullptr;
};
 
struct RtcpSchedulerContext {
                int intervalMs = 5000;
 
                int64_t wireSilenceTimeoutMs = 0;
 
                List<RtcpSchedulerWriterStream> writers;
 
                List<RtcpSchedulerReaderStream> readers;
 
                Function<void(RtcpSchedulerReaderStream &, int64_t gapNs)>
                        onWireSilenceEos;
};
 
class RtcpScheduler : public Thread {
        public:
                static constexpr unsigned int kStartupPollMs = 50;
 
                static constexpr int kDefaultIntervalMs = 5000;
 
                explicit RtcpScheduler(RtcpSchedulerContext ctx,
                                       const String        &name = String("rtp-rtcp"));
 
                ~RtcpScheduler() override;
 
                RtcpScheduler(const RtcpScheduler &) = delete;
                RtcpScheduler &operator=(const RtcpScheduler &) = delete;
 
                void requestStop();
 
                bool isStopRequested() const { return _stopRequested.value(); }
 
                void emitByeForAll();
 
                void runOnce();
 
        protected:
                void run() override;
 
        private:
                void emitOnce();
                void checkWireSilence(RtcpSchedulerReaderStream &s);
                bool allStreamsHaveEmitted() const;
                void cvSleep(unsigned int ms);
 
                static void emitForStream(RtcpSchedulerWriterStream &s);
                static void emitRrForStream(RtcpSchedulerReaderStream &s);
                static void emitByeForWriter(RtcpSchedulerWriterStream &s);
                static void emitByeForReader(RtcpSchedulerReaderStream &s);
 
                RtcpSchedulerContext _ctx;
                int                  _intervalMs;
                Atomic<bool>         _stopRequested;
                Mutex                _mutex;
                WaitCondition        _cv;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_NETWORK