Process.hpp

// SPDX-FileCopyrightText: 2025 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 <queue>
#include <unordered_map>
 
// Project includes
#include <dfx-core/Node.hpp>
#include <dfx-core/messages/DataMessage.hpp>
#include <dfx-fdwatch/Events.hpp>
#include <dfx-subprocess/Process.hpp>
#include <dfx-utilities/EnumString.hpp>
#include <dfx-utilities/FileSystem.hpp>
#include <dfx-utilities/MonotonicIdAllocator.hpp>
 
namespace dfx::Node
{
class Process : public Core::Node
{
    DFX_NODE("Process")
 
private:
    struct JsonParserState
    {
        bool hasEmitWarnForFirstNonValidChar = false;
        char openDelimiter = '\0';
        char closeDelimiter = '\0';
        uint32_t counter = 0;
        std::size_t idx = 0;
    };
 
    struct ChildInfo
    {
        ChildInfo(FdWatch::Poller & poller, Subprocess::Process::Config config)
            : process(poller, std::move(config))
        {}
 
        Subprocess::Process process;
        std::string stdoutData;
        std::string stderrData;
 
        JsonParserState stdoutParserState;
        JsonParserState stderrParserState;
    };
 
    using ChildInfos = std::unordered_map<uint32_t, ChildInfo>;
 
public:
    enum class LifeCycle
    {
        OneShot,
        Trigerred,
        TriggeredParallel,
        Daemon,
    };
 
    enum class EmitMode
    {
        Null,
        Line,
        Chunk,
        Whole,
        Json,
    };
 
    struct IO
    {
        EmitMode emitMode = EmitMode::Line;
    };
 
public:
    Process(Id id, std::string name);
    ~Process();
 
protected:
    void startImpl() override;
    void stopImpl() override;
 
    void initializeImpl(nlohmann::json config) override;
    void handleMessage(Core::InputPort const & port, Core::MessagePtr message) override;
 
private:
    ChildInfos::iterator _startNewChild();
 
    void _sendMessageToChild(ChildInfo & childInfo, Core::DataMessage const & message);
 
    void _onStdoutReadyRead(uint32_t id, std::string data);
    void _onStderrReadyRead(uint32_t id, std::string data);
    void _onChildTerminated(uint32_t id);
 
    std::vector<Core::MessagePtr> _buildMessages(Subprocess::Process::Which stream, ChildInfo & childInfo);
 
private:
    Subprocess::Process::Config _childConfig;
 
    LifeCycle _lifecycle;
    bool _shouldRestart = true;
 
    std::chrono::milliseconds _stopTimeout{0};
 
    std::queue<Core::MessagePtr> _stdinBuffer;
    IO _stdout;
    IO _stderr;
 
private:
    Utils::MonotonicIdAllocator<uint32_t> _childIdAllocator;
    ChildInfos _childInfos;
};
} // !namespace dfx::Node
 
DECLARE_ENUM_STRING_FUNCTIONS(dfx::Node::Process::LifeCycle);
DECLARE_ENUM_STRING_FUNCTIONS(dfx::Node::Process::EmitMode);