rtpdepacketizerthread.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/duration.h>
#include <promeki/error.h>
#include <promeki/namespace.h>
#include <promeki/queue.h>
#include <promeki/rtppacket.h>
#include <promeki/string.h>
#include <promeki/thread.h>
#include <promeki/timestamp.h>
 
PROMEKI_NAMESPACE_BEGIN
 
struct StreamAnchor {
                TimeStamp arrivalT0;
 
                uint32_t rtpTs0 = 0;
 
                uint32_t clockRate = 0;
 
                bool valid = false;
 
                void reset() {
                        *this = StreamAnchor{};
                }
 
                bool isValid() const { return valid && clockRate > 0; }
 
                TimeStamp captureTimeFor(uint32_t rtpTs) const {
                        if (!isValid()) return TimeStamp();
                        const uint32_t delta = rtpTs - rtpTs0;
                        const int64_t  ns = (static_cast<int64_t>(delta) * 1'000'000'000) /
                                            static_cast<int64_t>(clockRate);
                        return arrivalT0 + Duration::fromNanoseconds(ns);
                }
};
 
class RtpDepacketizerThread : public Thread {
        public:
                static constexpr size_t DefaultInputQueueDepth = 64;
 
                RtpDepacketizerThread(const String &name, uint32_t clockRateHz,
                                      size_t depth = DefaultInputQueueDepth);
 
                ~RtpDepacketizerThread() override;
 
                RtpDepacketizerThread(const RtpDepacketizerThread &) = delete;
                RtpDepacketizerThread &operator=(const RtpDepacketizerThread &) = delete;
 
                RtpPacket::Queue &inputQueue() { return _inputQueue; }
 
                const RtpPacket::Queue &inputQueue() const { return _inputQueue; }
 
                void requestStop();
 
                bool isStopRequested() const { return _stopRequested.value(); }
 
                uint32_t clockRateHz() const { return _clockRateHz; }
 
                const StreamAnchor &anchor() const { return _anchor; }
 
        protected:
                virtual void handlePacket(const RtpPacket &pkt) = 0;
 
                virtual void onStart() {}
 
                virtual void onStop() {}
 
                void ensureAnchor(uint32_t rtpTs, const TimeStamp &arrivalSteady);
 
                void resetAnchor();
 
                TimeStamp captureTimeForRtpTs(uint32_t rtpTs) const;
 
                static constexpr unsigned int PopTimeoutMs = 50;
 
                void run() override;
 
        private:
                Atomic<bool>      _stopRequested;
                RtpPacket::Queue  _inputQueue;
                uint32_t          _clockRateHz;
                StreamAnchor      _anchor;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_NETWORK