#include "message_bus/pipeline/PipelineEngine.h"

#include <chrono>

#include "utils/logs/logging.h"

namespace softbus::message_bus::pipeline
{

PipelineEngine::PipelineEngine(std::size_t queueCapacity)
    : m_ingressQ(queueCapacity), m_egressQ(queueCapacity)
{
}

PipelineEngine::~PipelineEngine()
{
    stop();
}

void PipelineEngine::start()
{
    if (m_running.exchange(true)) {
        return;
    }
    m_worker = std::thread([this]() { run(); });
}

void PipelineEngine::stop()
{
    if (!m_running.exchange(false)) {
        return;
    }
    if (m_worker.joinable()) {
        m_worker.join();
    }
}

void PipelineEngine::drain()
{
    PipelineContext tmp;
    while (m_ingressQ.pop(tmp)) {
    }
}

bool PipelineEngine::enqueueIngress(PipelineContext ctx)
{
    if (!m_ingressQ.push(std::move(ctx))) {
        ++m_counters.drop;
        return false;
    }
    ++m_counters.in;
    return true;
}

bool PipelineEngine::enqueueEgress(PipelineContext ctx)
{
    return m_egressQ.push(std::move(ctx));
}

bool PipelineEngine::tryPopEgress(PipelineContext& out)
{
    return m_egressQ.pop(out);
}

void PipelineEngine::setPluginManager(std::shared_ptr<softbus::core::plugin_system::PluginManager> pluginManager)
{
    m_pluginManager = std::move(pluginManager);
}

void PipelineEngine::run()
{
    while (m_running.load()) {
        PipelineContext ctx;
        if (!m_ingressQ.pop(ctx)) {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            continue;
        }

        const bool framed = stage1Framer(ctx);
        const bool parsed = framed && stage2Parser(ctx);
        const bool filtered = parsed && stage3Filter(ctx);
        const bool valid = filtered && stage4Validator(ctx);
        if (!valid) {
            ++m_counters.error;
            continue;
        }
        if (!m_egressQ.push(std::move(ctx))) {
            ++m_counters.drop;
            continue;
        }
        ++m_counters.out;
    }
}

bool PipelineEngine::stage1Framer(PipelineContext& ctx)
{
    if (!ctx.payload.valid()) {
        return false;
    }
    if (ctx.payloadSize == 0) {
        ctx.payloadSize = ctx.payload.length;
    }
    return true;
}

bool PipelineEngine::stage4Validator(PipelineContext& ctx)
{
    if (!ctx.payload.valid()) {
        return false;
    }
    if (ctx.endpoint.isEmpty()) {
        return false;
    }
    if (!ctx.timestamp.isValid()) {
        ctx.timestamp = QDateTime::currentDateTimeUtc();
    }
    return true;
}

bool PipelineEngine::stage2Parser(PipelineContext& ctx)
{
    if (!m_pluginManager) {
        return true;
    }
    auto plugin = m_pluginManager->selectProtocolPlugin(ctx.protocolHint);
    if (!plugin) {
        return true;
    }
    auto session = plugin->createSession();
    if (!session) {
        return false;
    }
    return session->feed(ctx);
}

bool PipelineEngine::stage3Filter(PipelineContext& ctx)
{
    (void)ctx;
    return true;
}

} // namespace softbus::message_bus::pipeline