medianode.h

Go to the documentation of this file.

 
#pragma once
 
#include <functional>
#include <mutex>
#include <promeki/core/namespace.h>
#include <promeki/core/objectbase.h>
#include <promeki/core/string.h>
#include <promeki/core/error.h>
#include <promeki/core/list.h>
#include <promeki/core/map.h>
#include <promeki/core/variant.h>
#include <promeki/core/queue.h>
#include <promeki/core/timestamp.h>
#include <promeki/proav/mediaport.h>
#include <promeki/proav/medialink.h>
#include <promeki/proav/frame.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class ThreadPool;
class MediaNode;
class MediaPipeline;
 
enum class Severity {
        Info,           
        Warning,        
        Error,          
        Fatal           
};
 
struct NodeMessage {
        Severity severity = Severity::Info;     
        String message;                         
        uint64_t frameNumber = 0;               
        TimeStamp timestamp;                    
        MediaNode *node = nullptr;              
};
 
struct NodeStats {
        uint64_t processCount = 0;              
        uint64_t starvationCount = 0;           
        double lastProcessDuration = 0.0;       
        double avgProcessDuration = 0.0;        
        double peakProcessDuration = 0.0;       
        int currentQueueDepth = 0;              
        int peakQueueDepth = 0;                 
};
 
class MediaNode : public ObjectBase {
        PROMEKI_OBJECT(MediaNode, ObjectBase)
        public:
                enum State {
                        Idle,           
                        Configured,     
                        Running,        
                        ErrorState      
                };
 
                enum ThreadingPolicy {
                        UseGraphPool,   
                        DedicatedThread,
                        CustomPool      
                };
 
                MediaNode(ObjectBase *parent = nullptr);
 
                virtual ~MediaNode();
 
                State state() const { return _state; }
 
                const String &name() const { return _name; }
 
                void setName(const String &name) { _name = name; return; }
 
                // ---- Port management ----
 
                const MediaPort::PtrList &inputPorts() const { return _inputPorts; }
 
                const MediaPort::PtrList &outputPorts() const { return _outputPorts; }
 
                MediaPort::Ptr inputPort(int index) const;
 
                MediaPort::Ptr outputPort(int index) const;
 
                MediaPort::Ptr inputPort(const String &name) const;
 
                MediaPort::Ptr outputPort(const String &name) const;
 
                int inputPortCount() const { return _inputPorts.size(); }
 
                int outputPortCount() const { return _outputPorts.size(); }
 
                // ---- Threading ----
 
                void setThreadingPolicy(ThreadingPolicy policy) {
                        _threadingPolicy = policy;
                        return;
                }
 
                void setThreadingPolicy(ThreadPool *pool) {
                        _threadingPolicy = CustomPool;
                        _customPool = pool;
                        return;
                }
 
                ThreadingPolicy threadingPolicy() const { return _threadingPolicy; }
 
                ThreadPool *customThreadPool() const { return _customPool; }
 
                // ---- Input queue ----
 
                void setIdealQueueSize(int size) { _idealQueueSize = size; return; }
 
                int idealQueueSize() const { return _idealQueueSize; }
 
                int queuedFrameCount() const { return _inputQueue.size(); }
 
                // ---- Virtual lifecycle ----
 
                virtual Error configure();
 
                virtual Error start();
 
                virtual void stop();
 
                virtual void process() = 0;
 
                virtual void starvation();
 
                // ---- Property interface ----
 
                virtual Map<String, Variant> properties() const;
 
                virtual Error setProperty(const String &name, const Variant &value);
 
                Variant property(const String &name) const;
 
                // ---- Node type registry ----
 
                static void registerNodeType(const String &typeName, std::function<MediaNode *()> factory);
 
                static MediaNode *createNode(const String &typeName);
 
                static List<String> registeredNodeTypes();
 
                // ---- Statistics ----
 
                NodeStats stats() const;
 
                void resetStats();
 
                virtual Map<String, Variant> extendedStats() const;
 
                // ---- Signals ----
 
                PROMEKI_SIGNAL(stateChanged, State);
 
                PROMEKI_SIGNAL(errorOccurred, Error);
 
                PROMEKI_SIGNAL(messageEmitted, NodeMessage);
 
        protected:
                void enqueueInput(Frame::Ptr frame);
 
                void recordProcessTiming(double duration);
 
                void recordStarvation();
 
                void addInputPort(MediaPort::Ptr port);
 
                void addOutputPort(MediaPort::Ptr port);
 
                void setState(State state);
 
                Frame::Ptr dequeueInput();
 
                void deliverOutput(int portIndex, Frame::Ptr frame);
 
                void deliverOutput(Frame::Ptr frame);
 
                void emitMessage(Severity severity, const String &message, uint64_t frameNumber = 0);
 
                void emitWarning(const String &message);
 
                void emitError(const String &message);
 
        private:
                String                  _name;
                State                   _state = Idle;
                ThreadingPolicy         _threadingPolicy = UseGraphPool;
                ThreadPool              *_customPool = nullptr;
                int                     _idealQueueSize = 2;
                MediaPort::PtrList      _inputPorts;
                MediaPort::PtrList      _outputPorts;
                Queue<Frame::Ptr>       _inputQueue;
 
                // Outgoing links (managed by MediaGraph)
                MediaLink::PtrList      _outgoingLinks;
                friend class MediaGraph;
                friend class MediaLink;
                friend class MediaPipeline;
 
                // Statistics (guarded by _statsMutex)
                mutable std::mutex      _statsMutex;
                uint64_t                _processCount = 0;
                uint64_t                _starvationCount = 0;
                double                  _lastProcessDuration = 0.0;
                double                  _avgProcessDuration = 0.0;
                double                  _peakProcessDuration = 0.0;
                int                     _peakQueueDepth = 0;
 
                static Map<String, std::function<MediaNode *()>> &nodeRegistry();
};
 
#define PROMEKI_REGISTER_NODE(ClassName) \
        static struct ClassName##Registrar { \
                ClassName##Registrar() { \
                        MediaNode::registerNodeType(#ClassName, []() -> MediaNode * { return new ClassName(); }); \
                } \
        } __##ClassName##Registrar;
 
PROMEKI_NAMESPACE_END