softbus_daemon/src/message_bus/pipeline/PipelineEngine.h

#pragma once
// 定义管道引擎 包含队列、运行状态、工作线程、计数器、插件管理器
#include <atomic>
#include <memory>
#include <thread>
#include <utility>
#include <vector>

#include <QHash>
#include <QJsonObject>
#include <QString>

#include "core/plugin_system/PluginManager.h"
#include "core/plugin_system/IProtocolFramerPlugin.h"
#include "core/plugin_system/IProtocolMapperPlugin.h"
#include "core/models/DOMMessage.h"
#include "core/models/EnrichedMessage.h"
#include "core/memory/LockFreeQueue.h"
#include "core/memory/MemoryPool.h"
#include "core/threading/MutexQueue.h"
#include "message_bus/pipeline/DynamicRuleRegistry.h"
#include "message_bus/pipeline/PipelineContext.h"
#include "message_bus/pipeline/stages/1_framers/PassthroughFramer.h"

namespace softbus::message_bus::pipeline
{

struct PipelineRuntimeConfig
{
    bool parallelPipeline{false};
    int  parseWorkerCount{2};
    bool orderedEgress{true};

    std::size_t framedQueueMax{4096};
    std::size_t maxReorderDepth{256};
};

struct EndpointFramingState;

class PipelineEngine
{
public:
    struct Counters
    {
        std::atomic<std::uint64_t> in{0};
        std::atomic<std::uint64_t> out{0};
        std::atomic<std::uint64_t> drop{0};
        std::atomic<std::uint64_t> error{0};
        std::atomic<std::uint64_t> mergeDrop{0};
    };

    explicit PipelineEngine(std::size_t queueCapacity = 4096);
    ~PipelineEngine();

    void start();
    void stop();
    void drain();

    bool enqueueIngress(PipelineContext ctx);
    bool enqueueEgress(PipelineContext ctx);

    bool tryPopEgress(PipelineContext& out);
    void setPluginManager(std::shared_ptr<softbus::core::plugin_system::PluginManager> pluginManager);
    void setMemoryPool(std::shared_ptr<softbus::core::memory::MemoryPool> pool);
    void setRuntimeConfig(PipelineRuntimeConfig cfg);
    void loadRuntimeConfigFromJson(const QJsonObject& obj);

    const Counters& counters() const { return m_counters; }

private:
    struct MergeItem
    {
        QString          endpoint;
        std::uint64_t    seq{0};
        PipelineContext ctx;
        bool             pipelineOk{false};
    };

    struct MergeState
    {
        std::uint64_t                                            nextExpected{1};
        QHash<std::uint64_t, std::pair<PipelineContext, bool>> buffer;
    };

    void runSingleWorker();
    void parseWorkerLoop();
    void mergerLoop();

    void ensureEndpointFramer(const QString& endpoint);

    void dispatchFramer(PipelineContext& chunkIn, std::vector<PipelineContext>& completeFramesOut);
    void assignFrameSequence(PipelineContext& ctx);
    bool runStages234(PipelineContext& ctx);
    void ingestMergeItem(MergeItem item);

    std::vector<softbus::core::models::DOMMessage> stage2Parser(const PipelineContext& ctx);
    bool stage3Filter(softbus::core::models::DOMMessage& message);
    bool stage4Validator(const softbus::core::models::DOMMessage& message) const;
    softbus::core::models::EnrichedMessage applyTransformAndEnrich(
        const PipelineContext& rawCtx, softbus::core::models::DOMMessage& message) const;

private:
    softbus::core::memory::LockFreeQueue<PipelineContext> m_ingressQ;
    softbus::core::memory::LockFreeQueue<PipelineContext> m_egressQ;
    std::atomic<bool> m_running{false};
    std::thread       m_worker;
    Counters          m_counters;
    std::shared_ptr<softbus::core::plugin_system::PluginManager> m_pluginManager;
    std::shared_ptr<softbus::core::memory::MemoryPool>           m_pool;

    PipelineRuntimeConfig m_rtConfig;

    std::shared_ptr<softbus::message_bus::pipeline::stages::framers::PassthroughFramer> m_passthroughFramer;
    std::mutex m_framerSessionMutex;
    QHash<QString, std::shared_ptr<softbus::core::plugin_system::IProtocolFramerSession>> m_framerSessions;
    std::mutex m_parserSessionMutex;
    QHash<QString, std::shared_ptr<softbus::core::plugin_system::IProtocolSession>> m_parserSessions;
    std::mutex m_mapperSessionMutex;
    QHash<QString, std::shared_ptr<softbus::core::plugin_system::IProtocolMapperSession>> m_mapperSessions;

    std::mutex                    m_seqMutex;
    QHash<QString, std::uint64_t> m_nextFrameSeqByEndpoint;

    std::unique_ptr<softbus::core::threading::MutexQueue<PipelineContext>> m_framedQ;
    std::unique_ptr<softbus::core::threading::MutexQueue<MergeItem>>       m_mergeQ;
    std::vector<std::thread>                                                m_parseThreads;
    std::thread                                                             m_mergerThread;

    std::mutex                                              m_endpointMutex;
    QHash<QString, std::shared_ptr<EndpointFramingState>> m_endpointFramers;

    std::mutex                 m_mergeStateMutex;
    QHash<QString, MergeState> m_mergeByEndpoint;
    int m_count{0}; // 计数器
};

} // namespace softbus::message_bus::pipeline