mediaioportconnection.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_PROAV
#include <functional>
 
#include <promeki/function.h>
#include <promeki/atomic.h>
#include <promeki/error.h>
#include <promeki/frame.h>
#include <promeki/framecount.h>
#include <promeki/list.h>
#include <promeki/mediaiorequest.h>
#include <promeki/namespace.h>
#include <promeki/objectbase.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class MediaIOSource;
class MediaIOSink;
 
class MediaIOPortConnection : public ObjectBase {
                PROMEKI_OBJECT(MediaIOPortConnection, ObjectBase)
        public:
                MediaIOPortConnection(MediaIOSource *source, MediaIOSink *sink, ObjectBase *parent = nullptr);
 
                explicit MediaIOPortConnection(MediaIOSource *source, ObjectBase *parent = nullptr);
 
                ~MediaIOPortConnection() override;
 
                MediaIOSource *source() const { return _source; }
 
                MediaIOSink *sink() const;
 
                List<MediaIOSink *> sinks() const;
 
                int sinkCount() const { return static_cast<int>(_sinks.size()); }
 
                bool isRunning() const { return _running; }
 
                Error addSink(MediaIOSink *sink, FrameCount frameLimit = FrameCount());
 
                Error start();
 
                void stop();
 
                // ---- Stats ----
 
                int64_t framesTransferred() const { return _framesTransferred.value(); }
 
                int64_t framesWritten(MediaIOSink *sink) const;
 
                bool upstreamDone() const { return _upstreamDone; }
 
                bool sinkStopped(MediaIOSink *sink) const;
 
                bool allSinksDone() const { return _allSinksDoneEmitted; }
 
                // ---- Per-sink gate (capture transport) ----
 
                void setSinkGate(MediaIOSink *sink, bool open);
 
                bool sinkGate(MediaIOSink *sink) const;
 
                using FrameInspector = Function<void(const Frame &)>;
 
                void setFrameInspector(FrameInspector cb);
 
                bool hasFrameInspector() const { return static_cast<bool>(_frameInspector); }
 
                // ---- Signals ----
 
                PROMEKI_SIGNAL(sinkGateChanged, MediaIOSink *, bool);
 
                PROMEKI_SIGNAL(upstreamDone);
 
                PROMEKI_SIGNAL(errorOccurred, Error);
 
                PROMEKI_SIGNAL(sinkLimitReached, MediaIOSink *);
 
                PROMEKI_SIGNAL(sinkError, MediaIOSink *, Error);
 
                PROMEKI_SIGNAL(allSinksDone);
 
                PROMEKI_SIGNAL(stopped);
 
        private:
                struct SinkState {
                                MediaIOSink *sink = nullptr;
                                FrameCount   frameLimit;
                                int64_t      framesWritten = 0;
                                bool         doneByLimit = false;
                                bool         stopped = false;
                                bool         gateOpen = true;
                                bool         pendingForceKeyframe = false;
                };
 
                void              pump();
                void              onSinkWriteError(MediaIOSink *sink, Error err);
                SinkState        *findSinkState(MediaIOSink *sink);
                const SinkState  *findSinkState(MediaIOSink *sink) const;
                void              maybeEmitAllSinksDone();
                bool              everyActiveSinkAcceptsWrite() const;
 
                void schedulePump();
 
                MediaIOSource           *_source = nullptr;
                List<SinkState> _sinks;
                FrameInspector           _frameInspector;
                // The pump holds the most recent in-flight read
                // request across iterations.  When @ref readFrame
                // returns a request whose payload has not yet
                // arrived (cache miss) the pump stashes it here and
                // returns; the next pump invocation (kicked by
                // @c frameReady) re-checks and consumes when ready,
                // so no prefetched frame is ever discarded.
                MediaIORequest           _pendingRead;
                bool                     _running = false;
                bool                     _upstreamDone = false;
                bool                     _allSinksDoneEmitted = false;
                Atomic<int64_t>          _framesTransferred{0};
                // Signal-coalescer: at most one pump is queued on
                // the loop at a time.  Many signals (frameReady +
                // per-sink frameWanted + per-cmd .then()) can fire
                // simultaneously from the strand worker; without
                // this flag they each post a pump callable and the
                // EventLoop's @c processEvents drains them faster
                // than @c pumpUntil can break out to check its
                // predicate.  The flag flips to true on the first
                // post and is cleared inside pump itself.
                Atomic<bool>             _pumpScheduled{false};
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_PROAV