GrabBag/App/StatorWorkpiecePosition/WorkpieceTaperedApp/Presenter/Src/TCPServerProtocol.cpp

#include "TCPServerProtocol.h"
#include "VrLog.h"
#include <QDateTime>
#include <QStringList>

namespace {
constexpr int kMaxClientBufferBytes = 64 * 1024;
}

TCPServerProtocol::TCPServerProtocol()
    : m_pTCPServer(nullptr)
    , m_bServerRunning(false)
    , m_nPort(7800)
    , m_connectionStatus(STATUS_DISCONNECTED)
{
}

TCPServerProtocol::~TCPServerProtocol()
{
    Deinitialize();
}

int TCPServerProtocol::Initialize(uint16_t port)
{
    LOG_DEBUG("Initializing TCP server protocol on port %d\n", port);

    m_nPort = port;

    // 创建TCP服务器实例
    if (!VrCreatYTCPServer(&m_pTCPServer)) {
        LOG_ERROR("Failed to create TCP server instance\n");
        return -1;
    }

    // 初始化TCP服务器
    if (!m_pTCPServer->Init(port)) {
        LOG_ERROR("Failed to initialize TCP server on port %d\n", port);
        delete m_pTCPServer;
        m_pTCPServer = nullptr;
        return -2;
    }

    // 设置事件回调
    m_pTCPServer->SetEventCallback([this](const TCPClient* pClient, TCPServerEventType eventType) {
        this->OnTCPEvent(pClient, eventType);
    });

    // Arm state before Start() creates its worker thread. A client can connect
    // immediately, so Initialize must not overwrite that event afterwards.
    m_connectionStatus.store(STATUS_DISCONNECTED);
    m_bServerRunning.store(true);

    // 启动TCP服务器
    if (!m_pTCPServer->Start([this](const TCPClient* pClient, const char* pData, const unsigned int nLen) {
        this->OnTCPDataReceived(pClient, pData, nLen);
    })) {
        LOG_ERROR("Failed to start TCP server\n");
        m_bServerRunning.store(false);
        IYTCPServer* failedServer = nullptr;
        {
            std::lock_guard<std::mutex> sendLock(m_sendMutex);
            failedServer = m_pTCPServer;
            m_pTCPServer = nullptr;
        }
        if (failedServer) {
            failedServer->Close();
            delete failedServer;
        }
        return -3;
    }

    LOG_DEBUG("TCP server protocol initialized successfully on port %d\n", port);
    return 0;
}

void TCPServerProtocol::Deinitialize()
{
    m_bServerRunning.store(false);

    // Detach the pointer while excluding result sends. New sends recheck the
    // flag under this mutex and return without touching the server pointer.
    // Do not hold the mutex while Stop() joins callbacks.
    IYTCPServer* server = nullptr;
    {
        std::lock_guard<std::mutex> sendLock(m_sendMutex);
        server = m_pTCPServer;
        m_pTCPServer = nullptr;
    }

    if (server) {
        LOG_DEBUG("Stopping TCP server protocol\n");

        server->Stop();
        server->Close();
        delete server;

        LOG_DEBUG("TCP server protocol stopped\n");
    }

    m_connectionStatus.store(ConnectionStatus::STATUS_DISCONNECTED);
    {
        std::lock_guard<std::mutex> buffersLock(m_clientBuffersMutex);
        m_connectedClients.clear();
        m_clientBuffers.clear();
    }
}

int TCPServerProtocol::SendTextFrame(const QByteArray& frame, const TCPClient* pClient)
{
    if (!m_bServerRunning.load()) {
        LOG_ERROR("TCP server is not running\n");
        return -1;
    }

    QByteArray payload = frame;
    while (payload.endsWith('\n') || payload.endsWith('\r')) {
        payload.chop(1);
    }
    payload.append("\r\n");

    std::lock_guard<std::mutex> sendLock(m_sendMutex);
    if (!m_pTCPServer || !m_bServerRunning.load()) {
        LOG_ERROR("TCP server is not running\n");
        return -1;
    }

    const bool success = pClient
        ? m_pTCPServer->SendData(pClient, payload.constData(), payload.size())
        : m_pTCPServer->SendAllData(payload.constData(), payload.size());
    if (!success) {
        LOG_ERROR("Failed to send TCP text frame\n");
        return -2;
    }

    LOG_DEBUG("Sent TCP text frame, size: %d bytes\n", payload.size());
    return 0;
}

TCPServerProtocol::TCPStatus TCPServerProtocol::GetConnectionStatus() const
{
    return m_connectionStatus.load();
}

void TCPServerProtocol::SetConnectionCallback(const ConnectionCallback& callback)
{
    m_connectionCallback = callback;
}

void TCPServerProtocol::SetDetectionTriggerCallback(const DetectionTriggerCallback& callback)
{
    m_detectionTriggerCallback = callback;
}

void TCPServerProtocol::SetBinDetectionTriggerCallback(const BinDetectionTriggerCallback& callback)
{
    m_binDetectionTriggerCallback = callback;
}

bool TCPServerProtocol::IsRunning() const
{
    return m_bServerRunning.load();
}

void TCPServerProtocol::OnTCPEvent(const TCPClient* pClient, TCPServerEventType eventType)
{
    if (!m_bServerRunning.load()) {
        return;
    }

    // CYTCPServer may report events from multiple worker tasks. Serialize the
    // aggregate-state transition and callback so an older event cannot publish
    // its result after a newer one.
    std::lock_guard<std::mutex> eventLock(m_connectionEventMutex);

    switch (eventType) {
        case TCP_EVENT_CLIENT_CONNECTED:
            LOG_DEBUG("TCP client connected: %p\n", pClient);
            {
                bool hasConnectedClient = false;
                {
                    std::lock_guard<std::mutex> buffersLock(m_clientBuffersMutex);
                    if (pClient) {
                        m_connectedClients.insert(pClient);
                        // Receive may race ahead of the connected callback. Do
                        // not overwrite a buffer created lazily by receive.
                        m_clientBuffers.emplace(pClient, QByteArray());
                    }
                    hasConnectedClient = !m_connectedClients.empty();
                }
                const TCPStatus newStatus = hasConnectedClient
                    ? STATUS_CONNECTED : STATUS_DISCONNECTED;
                if (m_connectionStatus.exchange(newStatus) != newStatus &&
                    m_connectionCallback) {
                    m_connectionCallback(hasConnectedClient);
                }
            }
            break;

        case TCP_EVENT_CLIENT_DISCONNECTED:
            LOG_DEBUG("TCP client disconnected: %p\n", pClient);
            {
                bool hasConnectedClient = false;
                {
                    std::lock_guard<std::mutex> buffersLock(m_clientBuffersMutex);
                    m_connectedClients.erase(pClient);
                    m_clientBuffers.erase(pClient);
                    hasConnectedClient = !m_connectedClients.empty();
                }
                const TCPStatus newStatus = hasConnectedClient
                    ? STATUS_CONNECTED : STATUS_DISCONNECTED;
                if (m_connectionStatus.exchange(newStatus) != newStatus &&
                    m_connectionCallback) {
                    m_connectionCallback(hasConnectedClient);
                }
            }
            break;

        case TCP_EVENT_CLIENT_EXCEPTION:
            LOG_WARNING("TCP client exception: %p\n", pClient);
            {
                bool hasConnectedClient = false;
                {
                    std::lock_guard<std::mutex> buffersLock(m_clientBuffersMutex);
                    m_connectedClients.erase(pClient);
                    m_clientBuffers.erase(pClient);
                    hasConnectedClient = !m_connectedClients.empty();
                }
                const TCPStatus newStatus = hasConnectedClient
                    ? STATUS_CONNECTED : STATUS_DISCONNECTED;
                if (m_connectionStatus.exchange(newStatus) != newStatus &&
                    m_connectionCallback) {
                    m_connectionCallback(hasConnectedClient);
                }
            }
            break;
    }
}

void TCPServerProtocol::OnTCPDataReceived(const TCPClient* pClient, const char* pData, unsigned int nLen)
{
    if (!m_bServerRunning.load()) {
        return;
    }

    if (!pClient || !pData || nLen == 0) {
        LOG_WARNING("Received empty data from client %p\n", pClient);
        return;
    }

    LOG_DEBUG("Received TCP data from client %p, size: %u bytes\n", pClient, nLen);

    bool bufferOverflow = false;
    std::vector<QByteArray> frames;

    {
        std::lock_guard<std::mutex> buffersLock(m_clientBuffersMutex);
        QByteArray& buffer = m_clientBuffers[pClient];

        if (nLen > static_cast<unsigned int>(kMaxClientBufferBytes) ||
            buffer.size() > kMaxClientBufferBytes - static_cast<int>(nLen)) {
            buffer.clear();
            bufferOverflow = true;
        } else {
            buffer.append(pData, static_cast<int>(nLen));

            while (true) {
                int idx = buffer.indexOf("\r\n");
                int terminatorLen = 2;
                if (idx < 0) {
                    idx = buffer.indexOf('\n');
                    terminatorLen = 1;
                }
                if (idx < 0) {
                    break;
                }

                QByteArray frame = buffer.left(idx).trimmed();
                buffer.remove(0, idx + terminatorLen);
                if (!frame.isEmpty()) {
                    frames.push_back(frame);
                }
            }
        }
    }

    if (bufferOverflow) {
        LOG_WARNING("TCP receive buffer exceeded 64 KiB for client %p\n", pClient);
        SendErrorResponse(pClient, -6, "命令长度超过64KiB", 0);
        return;
    }

    for (const QByteArray& frame : frames) {
        ParseTextCommand(pClient, frame);
    }
}

void TCPServerProtocol::ParseTextCommand(const TCPClient* pClient, const QByteArray& frame)
{
    const QString command = QString::fromUtf8(frame).trimmed();
    const QStringList tokens = command.split('_', QString::SkipEmptyParts);

    LOG_DEBUG("Received TCP text command: %s, tokens=%d\n",
              command.toStdString().c_str(), tokens.size());

    if (tokens.size() != 1) {
        LOG_WARNING("Invalid TCP text command, expected one command token: %s\n",
                    command.toStdString().c_str());
        SendErrorResponse(pClient, -3, "命令格式错误", QDateTime::currentMSecsSinceEpoch());
        return;
    }

    const QString typeToken = tokens[0].toUpper();
    const bool isWorkpieceDetection = typeToken.startsWith('S');
    const bool isBinDetection = typeToken.startsWith('T');
    if (!isWorkpieceDetection && !isBinDetection) {
        LOG_WARNING("Invalid TCP command type: %s\n", typeToken.toStdString().c_str());
        SendErrorResponse(pClient, -3, "命令类型错误", QDateTime::currentMSecsSinceEpoch());
        return;
    }

    bool ok = false;
    int cameraIndex = typeToken.mid(1).toInt(&ok);
    if (!ok || cameraIndex < 1) {
        cameraIndex = 1;
    }

    const qint64 timestamp = QDateTime::currentMSecsSinceEpoch();
    LOG_INFO("TCP trigger: type=%s, camera=%d\n",
             isWorkpieceDetection ? "S" : "T", cameraIndex);

    if (isWorkpieceDetection) {
        if (!m_detectionTriggerCallback) {
            SendErrorResponse(pClient, -4, "工件检测服务未准备就绪", timestamp);
            return;
        }
        if (!m_detectionTriggerCallback(true, cameraIndex, timestamp)) {
            SendErrorResponse(pClient, -5, "工件检测启动失败", timestamp);
        }
        return;
    }

    if (!m_binDetectionTriggerCallback) {
        SendErrorResponse(pClient, -4, "料框检测服务未准备就绪", timestamp);
        return;
    }
    if (!m_binDetectionTriggerCallback(cameraIndex, timestamp)) {
        SendErrorResponse(pClient, -5, "料框检测启动失败", timestamp);
    }
}

void TCPServerProtocol::SendErrorResponse(const TCPClient* pClient, int code, const QString& message, qint64 timestamp)
{
    (void)timestamp;
    const QString sanitizedMessage = QString(message).replace('_', '-').replace('\n', ' ');
    SendTextFrame(QString("Error_%1_%2").arg(code).arg(sanitizedMessage).toUtf8(), pClient);
}