Node.hpp

Go to the documentation of this file.

// SPDX-FileCopyrightText: 2025-2026 Vincent Leroy
// SPDX-License-Identifier: MIT
//
// This file is part of dfx.
//
// Licensed under the MIT License. See the LICENSE file in the project root
// for full license information.
 
#pragma once
 
// Standard includes
#include <concepts>
#include <future>
#include <mutex>
#include <string>
 
// Third-party includes
#include <nlohmann/json.hpp>
 
// Project includes
#include "ports/InputPort.hpp"
#include "ports/OutputPort.hpp"
#include <dfx-utilities/CompilerSupport.hpp>
#include <dfx-utilities/EnumString.hpp>
#include <dfx-utilities/MonotonicIdAllocator.hpp>
#include <dfx-utilities/StringMap.hpp>

 
namespace dfx::Graph
{
class NodeFactory;
} // !namespace dfx::Graph
 
namespace dfx::Runtime::Api
{
class NodeReactor;
class PollerProxy;
class NodeTaskExecutor;
using NodeTaskExecutorPtr = std::unique_ptr<NodeTaskExecutor>;
} // !namespace dfx::Runtime::Api
 
namespace dfx::FdWatch
{
class Poller;
} // !namespace dfx::FdWatch
 
namespace dfx::Core
{
class Node;
using NodePtr  = std::shared_ptr<Node>;
using NodeWPtr = std::weak_ptr<Node>;

template<typename T>
concept DerivedFromNode = std::derived_from<T, Node>;

class Node : public std::enable_shared_from_this<Node>
{
    friend class Graph::NodeFactory;
    friend class InputPort;
 
    using InputPortContainer  = std::vector<std::pair<std::string, InputPort>>;
    using OutputPortContainer = std::vector<std::pair<std::string, OutputPort>>;
 
public:
    enum class ExecutionFlowPolicy
    {
        AlwaysDispatch,    
        InlineIfAvailable, 
    };
 
public:
    using Id = uint32_t;
 
public:
    Node(std::string type, Id id, std::string name);
    virtual ~Node() = 0;


    std::string const & type() const noexcept                { return _type; }
    std::string const & name() const noexcept                { return _name; }
    Id id() const noexcept                                   { return _id; }
    nlohmann::json const & config() const noexcept           { return _config; }
    ExecutionFlowPolicy executionFlowPolicy() const noexcept { return _execPolicy; }
    bool isRunning() const noexcept                          { return _isRunning.load(std::memory_order::acquire); }
    bool allowsMimeTypePropagation() const noexcept          { return _allowsMimeTypePropagation; }

 
public:
    void setName(std::string newName);
 
public:
    template<DerivedFromNode T>
    bool is() const noexcept { return dynamic_cast<T const *>(this) != nullptr; }

    template<DerivedFromNode T>
    T & as() noexcept { return static_cast<T &>(*this); }

    template<DerivedFromNode T>
    T const & as() const noexcept { return static_cast<T const &>(*this); }
 
public:

    void start();

    void stop();

    void waitForStopped();

    bool stopRequested() const noexcept { return _stopRequested.test(std::memory_order::acquire); }

 
public:

    InputPortContainer const & inputPorts() const noexcept { return _inputPorts; }
    OutputPortContainer const & outputPorts() const noexcept { return _outputPorts; }

    bool hasInputPort(std::string_view portName) const noexcept
    { return _getInputPortItr(portName) != _inputPorts.end(); }

    bool hasOutputPort(std::string_view portName) const noexcept
    { return _getOutputPortItr(portName) != _outputPorts.end(); }

    bool hasInputPort(Port::Id id) const noexcept
    { return _getInputPortItr(id) != _inputPorts.end(); }

    bool hasOutputPort(Port::Id id) const noexcept
    { return _getOutputPortItr(id) != _outputPorts.end(); }

    InputPort const & inputPort(std::string_view portName) const;
    OutputPort const & outputPort(std::string_view portName) const;
    InputPort & inputPort(std::string_view portName);
    OutputPort & outputPort(std::string_view portName);

    InputPort const & inputPort(Port::Id id) const;
    OutputPort const & outputPort(Port::Id id) const;
    InputPort & inputPort(Port::Id id);
    OutputPort & outputPort(Port::Id id);

 
public:

    void sendMessage(Port::Id id, MessagePtr message);

    void sendMessage(std::string_view portName, MessagePtr message);

    void sendMessage(OutputPort & port, MessagePtr message);

 
public:

    void lock()              { _nodeLock.lock(); }
    void unlock()            { _nodeLock.unlock(); }
    bool try_lock() noexcept { return _nodeLock.try_lock(); }

 
public:
    static void validateNodeName(std::string_view name);
 
public:

    void setReactor(Runtime::Api::NodeReactor * reactor);

    Runtime::Api::NodeReactor * reactor() const noexcept { return _reactor; }

    FdWatch::Poller & poller();

    template<typename T = Runtime::Api::NodeTaskExecutor>
    T * customExecutor() const noexcept { return static_cast<T *>(_customExecutor.get()); }
 
protected:
    void assignTaskExecutor(Runtime::Api::NodeTaskExecutorPtr executor);

    void revokeTaskExecutor() noexcept;

 
public:

    virtual nlohmann::json configSchema() const = 0;
    virtual nlohmann::json metadata() const = 0;

 
protected:
    void initialize(nlohmann::json config);

    void setAllowsMimeTypePropagation(bool allowed) { _allowsMimeTypePropagation = allowed; }
 
protected:

    virtual void startImpl() {}
    virtual void stopImpl() { onNodeStopped(); }
    virtual void initializeImpl(nlohmann::json config) = 0;

    void onNodeStopped();

 
protected:

    OutputPort & registerOutputPort(std::string name, MimeType mimeType = MimeType::Any, std::optional<bool> allowsMimeTypePropagation = std::nullopt);

    InputPort & registerInputPort(std::string name, MimeTypes supportedMimeTypes = {}, std::optional<bool> allowsMimeTypePropagation = std::nullopt);

    OutputPort & registerControlOutputPort(std::string name);

    InputPort & registerControlInputPort(std::string name);

 
protected:
    virtual void setExecutionFlowPolicy(ExecutionFlowPolicy policy) { _execPolicy = policy; }

    virtual void handleMessage(InputPort const & port, MessagePtr message) = 0;
 
private:
    OutputPort & _registerOutputPort(std::string name, Kind kind, MimeType mimeType, std::optional<bool> allowsMimeTypePropagation);
    InputPort & _registerInputPort(std::string name, Kind kind, MimeTypes supportedMimeTypes, std::optional<bool> allowsMimeTypePropagation);
 
    InputPortContainer::iterator _getInputPortItr(std::string_view name) noexcept;
    InputPortContainer::const_iterator _getInputPortItr(std::string_view name) const noexcept;
    OutputPortContainer::iterator _getOutputPortItr(std::string_view name) noexcept;
    OutputPortContainer::const_iterator _getOutputPortItr(std::string_view name) const noexcept;
 
    InputPortContainer::iterator _getInputPortItr(Port::Id id) noexcept;
    InputPortContainer::const_iterator _getInputPortItr(Port::Id id) const noexcept;
    OutputPortContainer::iterator _getOutputPortItr(Port::Id id) noexcept;
    OutputPortContainer::const_iterator _getOutputPortItr(Port::Id id) const noexcept;
 
private:
    Id                       _id;
    std::string              _name;
    std::string              _type;
    nlohmann::json           _config;
    std::atomic_bool         _isRunning = false;
    std::atomic_flag         _stopRequested;
    std::promise<void>       _stopPromise;
    std::shared_future<void> _stopFuture;
    bool                     _allowsMimeTypePropagation = true;
 
    ExecutionFlowPolicy _execPolicy = ExecutionFlowPolicy::InlineIfAvailable;
 
    std::recursive_mutex _nodeLock;
    Runtime::Api::NodeReactor * _reactor = nullptr;
    std::unique_ptr<Runtime::Api::PollerProxy> _pollerProxy;
    Runtime::Api::NodeTaskExecutorPtr _customExecutor;
 
private:
    InputPortContainer  _inputPorts;
    OutputPortContainer _outputPorts;
 
private:
    Utils::MonotonicIdAllocator<Port::Id> _portIdAllocator;
};
} // !namespace dfx::Core

#define DFX_NODE(typeName) \
public: \
    static constexpr auto type = typeName; \
    \
public: \
    static nlohmann::json staticConfigSchema(); \
    nlohmann::json configSchema() const override { return staticConfigSchema(); } \
    static nlohmann::json staticMetadata(); \
    nlohmann::json metadata() const override { return staticMetadata(); }\
private:
 
DFX_DECLARE_ENUM_STRING_FUNCTIONS(dfx::Core::Node::ExecutionFlowPolicy);