mediapipeline.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/clock.h>
#include <promeki/duration.h>
#include <promeki/error.h>
#include <promeki/frame.h>
#include <promeki/list.h>
#include <promeki/logger.h>
#include <promeki/map.h>
#include <promeki/mediaio.h>
#include <promeki/mediaioportconnection.h>
#include <promeki/mediaioportgroup.h>
#include <promeki/mediapipelineconfig.h>
#include <promeki/mediapipelinestats.h>
#include <promeki/mediapipelinetrigger.h>
#include <promeki/mutex.h>
#include <promeki/namespace.h>
#include <promeki/objectbase.h>
#include <promeki/pipelineevent.h>
#include <promeki/set.h>
#include <promeki/string.h>
#include <promeki/stringlist.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class MediaIOStatsCollector;
 
class MediaPipeline : public ObjectBase {
                PROMEKI_OBJECT(MediaPipeline, ObjectBase)
        public:
                enum class State {
                        Empty,   
                        Built,   
                        Open,    
                        Running, 
                        Stopped, 
                        Closed   
                };
 
                enum class PlaybackState {
                        Idle,    
                        Playing, 
                        Paused,  
                        Seeking, 
                        Ended    
                };
 
                enum class CaptureState {
                        Idle,      
                        Armed,     
                        Recording, 
                        Paused     
                };
 
                explicit MediaPipeline(ObjectBase *parent = nullptr);
 
                ~MediaPipeline() override;
 
                // ------------------------------------------------------------
                // Lifecycle
                // ------------------------------------------------------------
 
                Error build(const MediaPipelineConfig &config, bool autoplan = false);
 
                Error injectStage(MediaIO *io);
 
                Error open();
 
                Error start();
 
                Error stop();
 
                Error close(bool block = true);
 
                static constexpr unsigned int DefaultCloseTimeoutMs = 3000;
 
                void setCloseTimeoutMs(unsigned int ms) { _closeTimeoutMs = ms; }
 
                unsigned int closeTimeoutMs() const { return _closeTimeoutMs; }
 
                bool isClosing() const { return _closing; }
 
                State state() const { return _state; }
 
                // ------------------------------------------------------------
                // Playback transport
                // ------------------------------------------------------------
 
                PlaybackState playbackState() const { return _playbackState; }
 
                Error play();
 
                Error pause();
 
                Error togglePlayPause();
 
                double rate() const;
 
                Error setRate(double r);
 
                FrameNumber currentFrame() const;
 
                Error seek(FrameNumber pos, MediaIOSeekMode mode = MediaIO_SeekDefault);
 
                Error stepForward(int64_t n = 1);
 
                Error stepBackward(int64_t n = 1);
 
                // ------------------------------------------------------------
                // Capture transport
                // ------------------------------------------------------------
 
                CaptureState captureState() const { return _captureState; }
 
                Error armCapture();
 
                Error startCapture();
 
                Error pauseCapture();
 
                Error resumeCapture();
 
                Error stopCapture();
 
                Error setCaptureTrigger(MediaPipelineTrigger::UPtr trig);
 
                Error setCaptureTrigger(Function<bool(const Frame &)> fn);
 
                Error setCaptureTrigger(const String &queryExpr);
 
                Error clearCaptureTrigger();
 
                // ------------------------------------------------------------
                // Introspection
                // ------------------------------------------------------------
 
                const MediaPipelineConfig &config() const { return _config; }
 
                MediaIO *stage(const String &name) const;
 
                StringList stageNames() const;
 
                StringList describe() const;
 
                MediaPipelineStats stats();
 
                MediaPipelineStageStats stageStats(const String &name);
 
                // ------------------------------------------------------------
                // Signals
                // ------------------------------------------------------------
 
                PROMEKI_SIGNAL(pipelineError, String, Error);
 
                PROMEKI_SIGNAL(stageOpened, String);
 
                PROMEKI_SIGNAL(stageStarted, String);
 
                PROMEKI_SIGNAL(stageStopped, String);
 
                PROMEKI_SIGNAL(stageClosed, String);
 
                PROMEKI_SIGNAL(finished, bool);
 
                PROMEKI_SIGNAL(closed, Error);
 
                PROMEKI_SIGNAL(statsUpdated, MediaPipelineStats);
 
                PROMEKI_SIGNAL(playbackStateChanged, PlaybackState);
 
                PROMEKI_SIGNAL(rateChanged, double);
 
                PROMEKI_SIGNAL(positionChanged, FrameNumber);
 
                PROMEKI_SIGNAL(captureStateChanged, CaptureState);
 
                // ------------------------------------------------------------
                // Event subscription
                // ------------------------------------------------------------
 
                using EventCallback = Function<void(const PipelineEvent &)>;
 
                int subscribe(EventCallback cb);
 
                void unsubscribe(int id);
 
                void setStatsInterval(Duration interval);
 
                Duration statsInterval() const { return _statsInterval; }
 
        private:
                // One record per outgoing route.  Back-pressure,
                // per-sink frame-count caps, and the actual frame
                // dispatch all live on the @ref MediaIOPortConnection
                // that owns this edge's source — the pipeline only
                // tracks enough state per edge to drive the cascade
                // close when the source EOFs, mark the cap as
                // observed, and look up the sink's stage name for
                // error reporting.
                struct EdgeState {
                                String       toName;
                                MediaIO     *to = nullptr;
                                MediaIOSink *toSink = nullptr;
                                bool         isSinkEdge = false;
                                bool         capReached = false;
                };
 
                // Each source-or-transit stage owns one SourceState entry.
                // Stages with no outgoing routes (pure sinks) do not appear
                // here — they participate only as a sink end of some
                // upstream stage's connection.  The @c connection field is
                // owned by the pipeline and lives from @ref start until
                // @ref close.
                struct SourceState {
                                MediaIO                 *from = nullptr;
                                MediaIOPortConnection   *connection = nullptr;
                                List<EdgeState> edges;
                                bool                     upstreamDone = false;
                };
 
                struct Subscriber {
                                int           id;
                                EventCallback fn;
                                EventLoop    *loop;
                };
 
                Error    destroyStages();
                Error    topologicallySort(List<String> &order) const;
                MediaIO *instantiateStage(const MediaPipelineConfig::Stage &s);
 
                void wireConnectionsForOpenedStage(const String &name);
 
                void publish(PipelineEvent ev);
 
                void publishStateChanged();
 
                void publishTransportStateChanged(const String &scope, const String &newState);
 
                void setPlaybackState(PlaybackState s);
 
                Error resolvePacingStage();
 
                Error resolveCaptureSinks();
 
                void setCaptureState(CaptureState s);
 
                void onCaptureFrame(const Frame &frame);
 
                void setCaptureGates(bool open);
 
                void installCaptureInspectorIfNeeded();
 
                void removeCaptureInspector();
 
                void publishStageState(const String &stageName, const String &transition);
 
                void installLoggerTap();
 
                void removeLoggerTap();
 
                void startStatsTimerIfNeeded();
 
                void stopStatsTimer();
 
                void emitStatsSnapshot();
 
                void onUpstreamDone(const String &srcName);
 
                void onSourceConnectionError(const String &srcName, Error err);
 
                void onSinkConnectionError(const String &srcName, MediaIOSink *sink, Error err);
 
                void onSinkLimitReached(const String &srcName, MediaIOSink *sink);
 
                void initiateClose(bool clean);
 
                void forceCloseRemaining();
 
                // ObjectBase override — routes the close watchdog's
                // TimerEvent to @ref forceCloseRemaining.
                void timerEvent(TimerEvent *e) override;
 
                void onStageClosed(const String &stageName, Error err);
 
                void finalizeClose();
 
                void attachStatsCollector(const String &name, MediaIO *io);
 
                void clearStatsCollectors();
 
                MediaPipelineStageStats buildStageStats(const String &name, MediaIO *io);
 
                MediaPipelineConfig                                  _config;
                State                                                _state = State::Empty;
                PlaybackState                                        _playbackState = PlaybackState::Idle;
                CaptureState                                         _captureState = CaptureState::Idle;
 
                // Pacing-stage resolution (Playback only).  Populated by
                // @ref resolvePacingStage during @ref open, cleared by
                // @ref destroyStages.  Null when the kind is Capture or
                // when no stage in the config flagged @c pacesPipeline=true.
                MediaIO          *_pacingStage = nullptr;
                MediaIOPortGroup *_pacingGroup = nullptr;
                Clock::Ptr        _pacingClock;
 
                // Capture-sink resolution (Capture only).  Populated by
                // @ref resolveCaptureSinks during @ref open, cleared by
                // @ref destroyStages.  Empty when the kind is Playback
                // or when no stage in the config flagged @c captureSink=true.
                List<MediaIO *>            _captureSinks;
                MediaPipelineTrigger::UPtr _captureTrigger;
                Map<String, MediaIO *>                      _stages;
                Map<String, MediaIO *>                      _injected;
                Map<String, MediaIOStatsCollector *>        _statsCollectors;
                Map<String, SourceState>                    _sources;
                List<String>                                _topoOrder;
 
                // Close-cascade bookkeeping.  Latched by
                // @ref initiateClose and unwound in
                // @ref finalizeClose.  @c _cleanFinish is the running
                // "no errors observed" bit that eventually feeds
                // @ref finishedSignal; drops to false on any
                // operational or close-time error.
                bool                 _closing = false;
                bool                 _cleanFinish = false;
                Set<String> _stagesAwaitingClosed;
                Error                _closeError = Error::Ok;
 
                // Close-watchdog timer.  Armed by @ref initiateClose,
                // stopped by @ref finalizeClose, and escalates to
                // @ref forceCloseRemaining when it fires.  Zero means
                // "no watchdog armed".
                unsigned int _closeTimeoutMs = DefaultCloseTimeoutMs;
                int          _closeWatchdogTimerId = -1;
 
                // Pipeline-wide frame-count cap, cached at build time
                // from @ref MediaPipelineConfig::frameCount.  Zero means
                // "no cap" (matches the @ref FrameCount::isFinite() gate).
                // Non-zero values are passed as the per-sink frame
                // limit to @ref MediaIOPortConnection::addSink for every
                // terminal-sink edge; the connection enforces the cap at
                // safe-cut-point boundaries and emits @c sinkLimitReached
                // when each sink hits it.  @ref _terminalSinksRemaining
                // counts the still-pending capped sinks; when it reaches
                // zero the pipeline initiates a clean close.
                int64_t _frameCountLimit = 0;
                int     _terminalSinksRemaining = 0;
 
                // Event subscription bookkeeping.  _subsMutex guards
                // _subscribers and the logger-listener handle so
                // subscribe / unsubscribe / publish can race with each
                // other across threads safely.  The actual dispatch in
                // publish() snapshots the list under the lock and
                // releases before invoking callbacks so a callback may
                // (un)subscribe re-entrantly without deadlocking.
                mutable Mutex                 _subsMutex;
                Map<int, Subscriber> _subscribers;
                int                           _nextSubId = 0;
                Logger::ListenerHandle        _loggerTap = 0;
 
                // Stats tick.  _statsInterval is the user-configured
                // wall-clock cadence; _statsTimerId is the active timer
                // id (negative when no timer is armed).
                Duration _statsInterval;
                int      _statsTimerId = -1;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_PROAV