ndimediaio.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_NDI
#include <promeki/namespace.h>
#include <promeki/atomic.h>
#include <promeki/audiobuffer.h>
#include <promeki/basicthread.h>
#include <promeki/audiodesc.h>
#include <promeki/audiomarker.h>
#include <promeki/clock.h>
#include <promeki/dedicatedthreadmediaio.h>
#include <promeki/framenumber.h>
#include <promeki/framerate.h>
#include <promeki/imagedesc.h>
#include <promeki/mediadesc.h>
#include <promeki/mediaiofactory.h>
#include <promeki/mediaioallocator.h>
#include <promeki/mediaiostats.h>
#include <promeki/metadata.h>
#include <promeki/mutex.h>
#include <promeki/ndiclock.h>
#include <promeki/pacinggate.h>
#include <promeki/pcmaudiopayload.h>
#include <promeki/queue.h>
#include <promeki/string.h>
#include <promeki/uncompressedvideopayload.h>
 
// Forward-declare the NDI SDK's opaque instance types so we can hold
// instance handles in the class body without dragging the full SDK
// headers into the public include.  The .cpp pulls in
// <Processing.NDI.Lib.h> to actually drive the API.
struct NDIlib_send_instance_type;
struct NDIlib_recv_instance_type;
 
PROMEKI_NAMESPACE_BEGIN
 
class NdiMediaIO : public DedicatedThreadMediaIO {
                PROMEKI_OBJECT(NdiMediaIO, DedicatedThreadMediaIO)
        public:
                static inline const MediaIOStats::ID StatsFramesSent{"NdiFramesSent"};
                static inline const MediaIOStats::ID StatsAudioFramesSent{"NdiAudioFramesSent"};
                static inline const MediaIOStats::ID StatsBytesSent{"NdiBytesSent"};
 
                static inline const MediaIOStats::ID StatsFramesReceived{"NdiFramesReceived"};
                static inline const MediaIOStats::ID StatsAudioFramesReceived{"NdiAudioFramesReceived"};
                static inline const MediaIOStats::ID StatsMetadataReceived{"NdiMetadataReceived"};
                static inline const MediaIOStats::ID StatsDroppedReceives{"NdiDroppedReceives"};
                static inline const MediaIOStats::ID StatsAudioSilenceFilled{"NdiAudioSilenceFilled"};
                static inline const MediaIOStats::ID StatsAudioGapEvents{"NdiAudioGapEvents"};
 
                static MediaIOAllocator::Ptr makePinnedHostAllocator();
 
                NdiMediaIO(ObjectBase *parent = nullptr);
 
                ~NdiMediaIO() override;
 
                int instanceID() const { return _instanceId; }
 
                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;
                void  cancelBlockingWork() override;
 
        private:
                Error openSink(const MediaIO::Config &cfg, const MediaDesc &mediaDesc);
                void  closeSink();
 
                Error openSource(const MediaIO::Config &cfg);
                void  closeSource();
                void  captureLoop();
 
                Error sendVideo(const UncompressedVideoPayload &vp);
                Error sendAudio(const PcmAudioPayload &ap);
 
                void ingestNdiAudio(int64_t timestampTicks, size_t samples, size_t channels,
                                    float rate, const uint8_t *planarFloatData,
                                    size_t channelStrideBytes);
 
                friend struct NdiMediaIOTestAccess;
 
                // NDI handles — opaque pointers to SDK-managed state.
                NDIlib_send_instance_type *_send = nullptr;
                NDIlib_recv_instance_type *_recv = nullptr;
 
                // Resolved at openSink() — used to pre-allocate the FLTP
                // conversion buffer once per stream rather than per frame.
                size_t _audioChannels   = 0;
                float  _audioSampleRate = 0.0f;
 
                // Configuration captured at open time.
                String _sendName;
                String _sendGroups;
                String _extraIps;
                bool   _sendClockVideo = false;
                bool   _sendClockAudio = false;
                bool   _sinkMode       = false;
 
                // Resolved video shape — populated at openSink time.
                ImageDesc _imageDesc;
                FrameRate _frameRate;
 
                // Telemetry — atomic so executeCmd(Stats) can read without
                // racing the strand worker that mutates these.
                Atomic<int64_t>     _framesSent{0};
                Atomic<int64_t>     _audioFramesSent{0};
                Atomic<int64_t>     _bytesSent{0};
 
                // ---- Source-mode state ----
                //
                // The capture thread loops on @c NDIlib_recv_capture_v3
                // with a short timeout, demultiplexes the returned
                // frames into the queues / ring below, and frees the
                // SDK buffers via the matching recv_free_* call.  All
                // queues are mutex-protected internally; the strand
                // drains them in @c executeCmd(MediaIOCommandRead).
                BasicThread          _captureThread;
                int                  _instanceId = 0; 
                Atomic<bool>         _stopFlag{false};
                Atomic<bool>         _readCancelled{false};
                // Reader-side video queue.  Capacity is small (matches
                // V4L2's "drop oldest, count drop") because NDI pushes
                // at the source's true frame rate and we want timing-
                // accurate delivery rather than a long backlog.
                static constexpr int     VideoQueueDepth = 2;
                Queue<UncompressedVideoPayload::Ptr> _videoQueue;
                AudioBuffer              _audioRing;
                Mutex                    _audioMutex;
                // Predicted timestamp for the next arriving sample,
                // computed as @c lastFrame.timestamp + samples * 1e7
                // / rate.  Stays in NDI 100ns ticks; guarded by
                // @ref _audioMutex.  The capture thread compares each
                // new frame's timestamp against this value to detect
                // gaps; the gap (if positive and within sanity
                // bounds) is bridged with
                // @ref AudioBuffer::pushSilence so the ring's sample
                // count and the sender's media timeline stay in
                // sync.  Zero when no prior frame has been pushed
                // since the last reset.
                //
                // The PTS of the first sample currently in the ring
                // is recovered directly from AudioBuffer's anchor
                // queue on drain — push() lays a sender-anchored
                // MediaTimeStamp per timestamped frame, so coalesced
                // multi-frame drains correctly report the oldest
                // still-buffered timestamp without any additional
                // tracking here.
                int64_t                  _audioNextSampleTicks  = 0;
                // Markers accumulated for the next drained payload.
                // Each entry's @c offset is the sample index within
                // the eventual payload (i.e. the value of
                // @c _audioRing.available() at the moment the marker
                // was appended); @c length is the sample count
                // covered.  The list is stamped onto the drained
                // payload's @ref Metadata::AudioMarkers in
                // executeCmd(Read) and then cleared.
                AudioMarkerList          _audioMarkersSinceDrain;
                // Total samples of silence injected by the gap-fill
                // logic since the last open.  Atomic so the strand's
                // executeCmd(Stats) can publish it without taking
                // @ref _audioMutex.  Mirrored as
                // @ref StatsAudioSilenceFilled.
                Atomic<int64_t>          _audioSilenceSamples{0};
                // Total contiguous silence-fill events since the
                // last open.  Mirrored as @ref StatsAudioGapEvents.
                Atomic<int64_t>          _audioGapEvents{0};
                Mutex                    _metadataMutex;
                Metadata                 _pendingMetadata;
                bool                     _hasPendingMetadata = false;
 
                // Reader configuration captured at openSource time.
                // The bit-depth tag controls how P216 frames are tagged
                // when emitted (see @ref NdiReceiveBitDepth).
                int                      _captureTimeoutMs = 100;
                int                      _bitDepthHint     = 0; // 0 = Auto (16-bit).
 
                // Reader-side telemetry.
                Atomic<int64_t>     _framesReceived{0};
                Atomic<int64_t>     _audioFramesReceived{0};
                Atomic<int64_t>     _metadataReceived{0};
                Atomic<int64_t>     _droppedReceives{0};
 
                // Source-mode clock — driven by NDI per-frame
                // timestamps from the capture thread.  Owned via
                // Clock::Ptr so the bound port-group can keep it
                // alive past close if any consumer still holds a
                // reference.
                Clock::Ptr _sourceClock;
 
                // External sink-mode pacing — one PacingGate per
                // stream so video and audio can drift independently
                // around the same clock (a single gate would conflate
                // their timelines).  Both gates share the same Clock
                // bound via executeCmd(MediaIOCommandSetClock).  The
                // gates' wait() runs on the dedicated worker thread
                // inside sendVideo / sendAudio (same thread as the
                // setter, so no synchronization required).  When no
                // clock is bound the gates are no-ops and the SDK's
                // internal clock_video / clock_audio flags remain in
                // control.
                PacingGate _videoGate;
                PacingGate _audioGate;
};
 
class NdiFactory : public MediaIOFactory {
        public:
                NdiFactory() = default;
 
                String name() const override { return String("Ndi"); }
                String displayName() const override { return String("NDI Stream"); }
                String description() const override {
                        return String("NDI (Network Device Interface) media transport "
                                      "(uncompressed video + audio + metadata over IP)");
                }
 
                bool canBeSource() const override { return true; }
                bool canBeSink() const override { return true; }
 
                StringList schemes() const override { return {String("ndi")}; }
                bool       canHandlePath(const String &path) const override;
                StringList enumerate() const override;
                Error      urlToConfig(const Url &url, Config *outConfig) const override;
 
                Config::SpecMap configSpecs() const override;
 
                MediaIO *create(const Config &config, ObjectBase *parent = nullptr) const override;
};
 
PROMEKI_NAMESPACE_END
 
 
#endif // PROMEKI_ENABLE_NDI