rtmpmediaio.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_NETWORK
#include <promeki/atomic.h>
#include <promeki/audiodesc.h>
#include <promeki/dedicatedthreadmediaio.h>
#include <promeki/enums_mediaio.h>
#include <promeki/enums_rtmp.h>
#include <promeki/frame.h>
#include <promeki/framenumber.h>
#include <promeki/framerate.h>
#include <promeki/histogram.h>
#include <promeki/imagedesc.h>
#include <promeki/mediaiofactory.h>
#include <promeki/namespace.h>
#include <promeki/pacinggate.h>
#include <promeki/queue.h>
#include <promeki/rtmpsession.h>
#include <promeki/string.h>
#include <promeki/uniqueptr.h>
#include <promeki/url.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class RtmpClient;
class Thread;
 
class RtmpMediaIO : public DedicatedThreadMediaIO {
                PROMEKI_OBJECT(RtmpMediaIO, DedicatedThreadMediaIO)
        public:
                // ---- Telemetry keys ----
 
                static inline const MediaIOStats::ID StatsFramesSent{"FramesSent"};
                static inline const MediaIOStats::ID StatsFramesReceived{"FramesReceived"};
                static inline const MediaIOStats::ID StatsVideoMessagesSent{"VideoMessagesSent"};
                static inline const MediaIOStats::ID StatsAudioMessagesSent{"AudioMessagesSent"};
                static inline const MediaIOStats::ID StatsVideoMessagesReceived{"VideoMessagesReceived"};
                static inline const MediaIOStats::ID StatsAudioMessagesReceived{"AudioMessagesReceived"};
                static inline const MediaIOStats::ID StatsBytesSent{"BytesSent"};
                static inline const MediaIOStats::ID StatsBytesReceived{"BytesReceived"};
                static inline const MediaIOStats::ID StatsSendQueueDepth{"SendQueueDepth"};
                static inline const MediaIOStats::ID StatsReadQueueDepth{"ReadQueueDepth"};
                static inline const MediaIOStats::ID StatsSendQueueOverflows{"SendQueueOverflows"};
                static inline const MediaIOStats::ID StatsConnectDurationMs{"ConnectDurationMs"};
                static inline const MediaIOStats::ID StatsHandshakeDurationMs{"HandshakeDurationMs"};
                static inline const MediaIOStats::ID StatsVideoFramesDroppedPreIdr{"VideoFramesDroppedPreIdr"};
 
                static inline const MediaIOStats::ID StatsPacingTicksOnTime{"PacingTicksOnTime"};
                static inline const MediaIOStats::ID StatsPacingTicksLate{"PacingTicksLate"};
                static inline const MediaIOStats::ID StatsPacingTicksSkipped{"PacingTicksSkipped"};
                static inline const MediaIOStats::ID StatsPacingReanchors{"PacingReanchors"};
                static inline const MediaIOStats::ID StatsPacingClockKind{"PacingClockKind"};
 
                explicit RtmpMediaIO(ObjectBase *parent = nullptr);
 
                ~RtmpMediaIO() override;
 
                uint64_t objectId() const { return _objectId; }
 
                Error proposeInput(const MediaDesc &offered, MediaDesc *preferred) const override;
 
        protected:
                Error executeCmd(MediaIOCommandOpen &cmd) override;
                Error executeCmd(MediaIOCommandClose &cmd) override;
                Error executeCmd(MediaIOCommandRead &cmd) override;
                Error executeCmd(MediaIOCommandWrite &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 the reader queue pop.
                void cancelBlockingWork() override;
 
        private:
                class PacketizerThread;       
                class DepacketizerThread;     
 
                void resetAll();
 
                RtmpConnectOptions buildConnectOptions(const MediaIO::Config &cfg) const;
 
                static bool hasVideoEssence(const Frame &frame);
 
                static bool hasAudioEssence(const Frame &frame);
 
                bool paceVideoFrame();
 
                void armVideoPaceGate();
 
                String paceClockKind() const;
 
                void onClientDisconnected(Error reason);
 
                // Owned RtmpClient — instantiated in executeCmd(Open),
                // destroyed in resetAll.  Lives across the open/close
                // cycle so the writer / reader threads it owns service
                // the wire.
                UniquePtr<RtmpClient> _client;
 
                // Worker threads.  Both are nullptr when idle.
                UniquePtr<PacketizerThread>   _packetizer;
                UniquePtr<DepacketizerThread> _depacketizer;
 
                // Strand-side bounded reader queue — populated by the
                // depacketizer thread, drained by executeCmd(Read).
                Queue<Frame> _readerQueue;
 
                // Cancellation latch for executeCmd(Read).  Set by
                // cancelBlockingWork(); cleared at every Open.
                Atomic<bool>      _readCancelled{false};
 
                // Latched when @c RtmpClient::disconnectedSignal fires
                // (peer went away, socket I/O error, etc.).  Polled by
                // the packetizer / depacketizer worker loops so they
                // exit cleanly instead of spewing per-frame send
                // failures into the log, and consulted by
                // executeCmd(Read) / executeCmd(Write) so the failure
                // surfaces to the pipeline as a write/read error.
                // @c _disconnectErrorCode stores the disconnect reason
                // as an @c Error::Code (int) to keep the atomic POD;
                // reconstruct an @c Error via @c Error(static_cast).
                Atomic<bool>      _clientDisconnected{false};
                Atomic<int>       _disconnectErrorCode{0};
 
                // Sink-side state set at Open from the MediaConfig +
                // pending descriptor, read by the packetizer thread.
                ImageDesc _imageDesc;
                AudioDesc _audioDesc;
                Url       _url;
                String    _streamKey;
                bool      _readerMode = false;
                bool      _dropUntilKeyframe = true;
                bool      _repeatParameterSets = true;
                bool      _enhancedRtmp = true;
                bool      _emitAnnexB = false;
                bool      _dataEnabled = true;
                int       _sendQueueDepth = 64;
                int       _readQueueDepth = 64;
 
                // Per-instance frame counters used to populate
                // currentFrame / frameCount on read / write commands.
                // Strand-owned; updated only inside executeCmd().
                FrameCount _frameCount{0};
                FrameCount _framesSent{0};
 
                // Cumulative stats counters bumped by the worker
                // threads, read by executeCmd(Stats).  Atomics for
                // lock-free cross-thread aggregation.
                Atomic<int64_t> _readerFramesReceived{0};
                Atomic<int64_t> _videoFramesDroppedPreIdr{0};
                Atomic<int64_t> _sendQueueOverflows{0};
 
                // Strand-side video pacing.  Mirrors RtpMediaIO's
                // _videoGate but defaults to an Internal wall-clock
                // binding since TCP has no kernel-pacing analog to fall
                // back on.  Touched only from the strand.
                RtmpVideoPacing _videoPacingMode{RtmpVideoPacing::Internal};
                PacingGate      _videoPaceGate;
                FrameRate       _frameRate;
                int             _paceSkipThresholdMs = 0;
                int             _paceReanchorThresholdMs = 0;
                bool            _paceClockIsExternal = false;
 
                // Open-time telemetry — captured around the
                // RtmpClient::open / publish calls.
                int64_t _connectDurationMs = 0;
                int64_t _handshakeDurationMs = 0;
 
                // Process-local monotonic counter that gives every
                // RtmpMediaIO a distinct id within one process.
                static Atomic<uint64_t> _nextObjectId;
                uint64_t                _objectId = 0;
};
 
class RtmpFactory : public MediaIOFactory {
        public:
                RtmpFactory() = default;
 
                String name() const override { return String("Rtmp"); }
                String displayName() const override { return String("RTMP Stream"); }
                String description() const override {
                        return String("RTMP / RTMPS publisher and subscriber "
                                      "(H.264 + HEVC video, AAC audio)");
                }
                StringList schemes() const override {
                        return StringList{String("rtmp"), String("rtmps")};
                }
 
                bool canBeSource() const override { return true; }
                bool canBeSink() const override { return true; }
 
                Config::SpecMap configSpecs() const override;
                Error           urlToConfig(const Url &url, Config *outConfig) const override;
                MediaIO        *create(const Config &config, ObjectBase *parent = nullptr) const override;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_NETWORK