eUi/rpcModule/rpc_finsEip/rpc_finsEip.cpp

#include "rpc_finsEip.h"
#include "qprocess.h"
#include <RpcExeProcessThread.h>
#include <QHostAddress>
#include <QVariant>
#include <QDebug>

#include <QTcpSocket>
#include <QJsonObject>
#include <QJsonDocument>
#include <windows.h>
#include <QDateTime>
#include <QCoreApplication>
#include <QDir>
#include <QFileInfo>
#include <QThread>
#include <QTimer>
#include <QCryptographicHash>
#include <QScopedPointer>
#include <QBuffer>
#include <QPointer>
#include <QRegularExpression>
#include <QDataStream>
#include <QElapsedTimer>
#include <QMetaObject>
#include <QMutexLocker>
#include <algorithm>
#include <memory>
#include <cmath>
#include <cstring>
#include <atomic>

static RpcExeProcessThread* processThread = nullptr;
void stopProcess();

namespace {
template<typename Worker>
void stopWorkerThread(QThread*& thread, Worker*& worker, const char* name)
{
    if (thread == nullptr) {
        worker = nullptr;
        return;
    }

    if (thread->isRunning()) {
        thread->requestInterruption();
        thread->quit();
        if (!thread->wait(5000)) {
            qWarning() << name << "quit timeout; terminate as last resort before deleting QThread";
            thread->terminate();
            if (!thread->wait(2000)) {
                qWarning() << name << "is still running; skip deleting QThread to avoid Qt warning";
                thread->setParent(nullptr);
                worker = nullptr;
                return;
            }
        }
    }

    delete thread;
    thread = nullptr;
    worker = nullptr;
}
}

rpc_finsTcp* create(){
	qDebug()  << "rpc_finsTcp create success!";
	return new rpc_finsTcp;
}

void destory(rpc_finsTcp* ptr){

	if (ptr)
	{
		delete  ptr;
		ptr = nullptr;
	}
	qDebug()  << " rpc_finsTcp destory success!";
}

rpc_finsTcp::rpc_finsTcp(){

}

rpc_finsTcp::~rpc_finsTcp() {
    qDeleteAll(read_JsonRpcClient);
    read_JsonRpcClient.clear();

    for (auto& client : write_JsonRpcClients) {
        if (client.second) {
            QObject::disconnect(this, nullptr, client.second, nullptr);
        }
        stopWorkerThread(client.first, client.second, "JsonRpcClient worker thread");
    }
    if (hearBeatThread) {
        if (hearBeatClinet) {
            QObject::disconnect(this, nullptr, hearBeatClinet, nullptr);
        }
        stopWorkerThread(hearBeatThread, hearBeatClinet, "heartbeat JsonRpcClient thread");
    }
    stopProcess();
    qDebug() << " ~rpc_finsTcp success!";
}
QQueue<QByteArray> readDataQueue;
// 从类中读数据到实例对象
QByteArray variantListToByteArray(const QVariant& var) {
    QByteArray byteArray;

    if (var.canConvert<QVariantList>()) {
        QVariantList list = var.toList();
        byteArray.reserve(list.size());

        for (const QVariant& item : list) {
            // 每个元素转为 uchar 存入 QByteArray
            byteArray.append(static_cast<char>(item.toInt()));
        }
    }

    return byteArray;
}
bool rpc_finsTcp::readData(std::vector<cv::Mat>& vec_mat, QByteArray& data){
    JsonRpcClient* ptr = read_JsonRpcClient[data];
    if(ptr == nullptr) {
        ptr = new JsonRpcClient("127.0.0.1", 9000);
        ptr->writeCall("init", QJsonArray{ m_IP,m_port }, "eiptcp_read");
        read_JsonRpcClient[data] = ptr;
	}
    QByteArray resp = ptr->readCall("excuteCli", QString(data), "eiptcp_read");
    data = resp; // 已是解码后的原始字节
	return true;
}

QByteArray rpc_finsTcp::executeReadCommand(QByteArray& input) {
    JsonRpcClient* ptr = read_JsonRpcClient[input];
    if (ptr == nullptr) {
        ptr = new JsonRpcClient("127.0.0.1", 9000);
        ptr->writeCall("init", QJsonArray{ m_IP, m_port }, "eiptcp_read");
        read_JsonRpcClient[input] = ptr;
    }
    return ptr->readCall("excuteCli", QString(input), "eiptcp_read");
}

bool rpc_finsTcp::executeWriteCommand(QByteArray& input) {
    if (input.contains("null"))
        return true;

    if (input.startsWith("init_")) {
        return parseInitCommand(input);
    }

    QList<QByteArray> m_list = input.split('_');
    if (m_list.size() < 4) {
        qCritical() << input + " m_list.size() != 4;command err";
        return false;
    }

    QString commond = m_list.at(0);
    if (commond == "read") {
        qWarning() << "Read command was sent to executeWriteCommand:" << input;
        return false;
    }

    if (write_JsonRpcClients.empty() && !input.contains("Heart")) {
        qCritical() << "RPC write clients are not initialized.";
        return false;
    }

    if (input.contains("Heart")) {
        auto request = packageWriteRequest("excuteCli", QJsonArray{ QString(input) }, "eiptcp_write");
        emit writeCallSignal(request.requestByte, request.id, input, -1);
        return true;
    }

    int threadIndex = 0;
    {
        QMutexLocker locker(&threadUseIndexMutex);
        threadIndex = threadUseIndex;
        if (++threadUseIndex >= static_cast<int>(write_JsonRpcClients.size())) threadUseIndex = 0;
    }

    auto request = packageWriteRequest("excuteCli", QJsonArray{ QString(input) }, "eiptcp_write");
    emit writeCallSignal(request.requestByte, request.id, input, threadIndex);
    return true;
}

bool rpc_finsTcp::writeData(std::vector<cv::Mat>& vec_mat, QByteArray& data) {
    QList<QByteArray> m_list = data.split('_');
    if (m_list.size()   )
    {
        qCritical() << data + " m_list.size() != 3;command err";
        return false;
    }
    QString commond = m_list.at(0);
    if (commond == "init") {
        m_IP = m_list.at(1);
        m_port = m_list.at(2).toInt();
        //threadCount = m_list.at(3).toInt();
    }
    else if (data.contains("read")) {
        readDataQueue.append(data);
    }
    else {
        if (data.contains("null"))
            return true;
        if (data.contains("Heart") == true) {//专门的发送心跳线程去发送此数据
            auto request = packageWriteRequest("excuteCli", QJsonArray{ QString(data) }, "eiptcp_write");
            emit writeCallSignal(request.requestByte, request.id,data, -1);
            return true;
        }
        //检查是不是复数命令
        QList<QByteArray> commands = data.split('#');

        int threadIndex = -1;
        {
            QMutexLocker locker(&threadUseIndexMutex);
            threadIndex = threadUseIndex;
            if (++threadUseIndex >= static_cast<int>(write_JsonRpcClients.size())) threadUseIndex = 0;
        }
        auto request = packageWriteRequest("excuteCli", QJsonArray{ QString(data) }, "eiptcp_write");
        emit writeCallSignal(request.requestByte, request.id, data, threadIndex);

        //
    }
    return true;
}


void startProcess(const QString& exePath,bool enableLog = false) {//注意：日志文件很大
    if (!processThread) {
        processThread = new RpcExeProcessThread();
    }
    processThread->startExe(exePath, enableLog ? QStringList{ "-l" } : QStringList{});
}

void stopProcess() {
    if (processThread) {
        processThread->stopExe();
        if (processThread->isRunning() && !processThread->wait(2000)) {
            qWarning() << "RPC process thread stop timeout; terminate as last resort";
            processThread->terminate();
            processThread->wait(1000);
        }
        if (processThread->isRunning()) {
            qWarning() << "RPC process thread is still running; skip deleting QThread";
            processThread = nullptr;
            return;
        }
        delete processThread;
        processThread = nullptr;
    }
}

void rpc_finsTcp::initClientAndThread(std::pair<QThread*, JsonRpcClientWorker*>& pairData,const int& i) {
    pairData.first = new QThread();
    pairData.second = new JsonRpcClientWorker(i);
    pairData.second->moveToThread(pairData.first);
    QObject::connect(pairData.first, &QThread::finished, pairData.second, &QObject::deleteLater);

    QObject::connect(this, &rpc_finsTcp::writeCallSignal,
        pairData.second, &JsonRpcClientWorker::writeCallSlot, Qt::QueuedConnection);
    pairData.first->start();

    QMetaObject::invokeMethod(pairData.second, "initClient", Qt::BlockingQueuedConnection);

    if (pairData.second->connection == Disconnected) {
        qDebug() << "JsonRpcClientSubThread connection failed! threadId" << i;
    }
    else if (pairData.second->connection == Connected) {
        qDebug() << "JsonRpcClientSubThread connected successfully! threadId" << i;
    }
}
void rpc_finsTcp::initHearBeatThread() {
    hearBeatThread = new QThread();
    hearBeatClinet = new JsonRpcClientWorker(-1);
    hearBeatClinet->moveToThread(hearBeatThread);
    QObject::connect(hearBeatThread, &QThread::finished, hearBeatClinet, &QObject::deleteLater);

    QObject::connect(this, &rpc_finsTcp::writeCallSignal,
        hearBeatClinet, &JsonRpcClientWorker::writeCallSlot, Qt::QueuedConnection);
    hearBeatThread->start();
    QMetaObject::invokeMethod(hearBeatClinet, "initClient", Qt::BlockingQueuedConnection);

    if (hearBeatClinet->connection == Disconnected) {
        qDebug() << "JsonRpcClientSubThread connection failed! heartBeatThread";
    }
    else if (hearBeatClinet->connection == Connected) {
        qDebug() << "JsonRpcClientSubThread connected successfully! threadId heartBeatThread";
    }
}

bool rpc_finsTcp::parseInitCommand(const QByteArray& input) {
    QList<QByteArray> m_list = input.split('_');
    if (m_list.size() < 3 || m_list.at(0) != "init") {
        qCritical() << input + " init command format error, expected init_ip_port_threadCount";
        return false;
    }

    bool portOk = false;
    //bool threadCountOk = false;
    const int parsedPort = m_list.at(2).toInt(&portOk);
    //const int parsedThreadCount = m_list.at(3).toInt(&threadCountOk);
    if (!portOk  ) {
        qCritical() << input + " init command value error";
        return false;
    }

    m_IP = m_list.at(1);
    m_port = parsedPort;
    //threadCount = parsedThreadCount;
    return true;
}

bool rpc_finsTcp::sendInitCommand(QString moduleName) {
    if (m_IP.isEmpty() || m_port <= 0) {
        qCritical() << "RPC init parameters are invalid:" << m_IP << m_port;
        return false;
    }

    // 初始化必须同步确认。此前通过 threadId=-1 的异步信号发送，而心跳
    // worker 尚未创建时该信号会被所有写 worker 忽略，且 PLC 连接失败也会被
    // 当作初始化成功。
    JsonRpcClient initClient("127.0.0.1", 9000);
    if (initClient._connectionStatus != Connected) {
        qCritical() << "Cannot connect JSON-RPC server while initializing" << moduleName;
        return false;
    }

    const QVariant result = initClient.call("init", QJsonArray{ m_IP, m_port }, moduleName);
    if (!result.isValid() || result.toString() != "OK") {
        qCritical() << "PLC initialization failed for" << moduleName << ":" << result;
        return false;
    }
    return true;
}

bool rpc_finsTcp::initParas(QByteArray& data){
    if (!parseInitCommand(data)) {
        return false;
    }

    startProcess("./rpc_server_Eip_1_0_6.exe");
    QThread::msleep(500);

	bool initSucess = true;
    constexpr int writeClientCount = 1;
    write_JsonRpcClients = std::vector<std::pair<QThread*, JsonRpcClientWorker*>>(writeClientCount);
    for (int i = 0; i < write_JsonRpcClients.size(); i++) {
        initClientAndThread(write_JsonRpcClients[i],i);
    }

    // 先创建 -1 对应的心跳 worker，避免后续心跳请求没有接收者。
    initHearBeatThread();

    if (!sendInitCommand("eiptcp_read"))
        initSucess = false;
    if (!sendInitCommand("eiptcp_write"))
        initSucess = false;
    return initSucess;

}

bool rpc_finsTcp::run(){
	// 没有用到
	//qDebug() << "[info] " << __FUNCTION__ << " line:" << __LINE__ << " success!";
	return true;
}


JsonRpcClient::JsonRpcClient(const QString& host, quint16 port)
    :host_(host), port_(port), idCounter_(1) {

    socket_.connectToHost(host_, port_);
    if (!socket_.waitForConnected(3000)) {
        qWarning() << "[JsonRpcClient] Initial connection failed:" << socket_.errorString();
        _connectionStatus = Disconnected;
    }
    else {
        qDebug() << "[JsonRpcClient] Connected to" << host_ << ":" << port_;
        _connectionStatus = Connected;
    }
}

bool JsonRpcClient::reconnect() {
    if (socket_.state() == QAbstractSocket::ConnectedState) {
        return true;
    }

    socket_.abort();  // 清理旧连接
    socket_.connectToHost(host_, port_);
    if (socket_.waitForConnected(3000)) {
        qDebug() << "[JsonRpcClient] Reconnected successfully.";
        return true;
    }
    else {
        qWarning() << "[JsonRpcClient] Reconnection failed:" << socket_.errorString();
        return false;
    }
}
QVariant JsonRpcClient::call(const QString& method, const QJsonArray& params, const QString& module){
    QMutexLocker locker(&mutex_);
    if (socket_.state() != QAbstractSocket::ConnectedState) {
        if (!reconnect()) return QVariant();
    }
    int myId = idCounter_++;
    QJsonObject req{
        {"jsonrpc", "2.0"},
        {"method", method},
        {"params", params},
        {"id", myId}
    };
    if (!module.isEmpty()) req.insert("module", module);

    QJsonObject resp = sendRequestBuildAndAwait(req);
    if (resp.contains("error")) {
        qWarning() << "[JsonRpcClient] Error:" << QJsonDocument(resp["error"].toObject()).toJson(QJsonDocument::Compact);
        return QVariant();
    }
    if (!resp.contains("result")) return QVariant();
    QJsonValue r = resp.value("result");
    if (r.isString()) return r.toString();
    if (r.isDouble()) return r.toDouble();
    if (r.isBool()) return r.toBool();
    if (r.isArray()) return r.toArray().toVariantList();
    if (r.isObject()) return r.toObject().toVariantMap();
    return QVariant();
}

QByteArray JsonRpcClient::readCall(const QString& method, const QString& params, const QString& module) {
    auto commands = params.split("_");
    QMutexLocker locker(&mutex_);
    if (socket_.state() != QAbstractSocket::ConnectedState) {
        if (!reconnect()) return QByteArray();
    }
    int myId = idCounter_++;
    QJsonArray p; p.append(params);
    QJsonObject req{{"jsonrpc","2.0"},{"method",method},{"params",p},{"id",myId}};
    if (!module.isEmpty()) req.insert("module", module);

    QJsonDocument doc(req);
    QByteArray payload = doc.toJson(QJsonDocument::Compact);
    if (socket_.write(payload) == -1 || !socket_.waitForBytesWritten(1000)) {
        qWarning() << "[JsonRpcClient] Write failed.";
        return QByteArray();
    }

    QJsonObject respObj;
    if (!waitAndReadResponse(myId, respObj)) {
        qWarning() << "[JsonRpcClient] Read timeout or parse error.";
        return QByteArray();
    }
    QString b64 = respObj.value("result").toString();
    QByteArray decoded = QByteArray::fromBase64(b64.toUtf8());
    if (b64.contains("error") || decoded.size()% commands[3].toInt() != 0) {
        qWarning() << b64;
        QThread::msleep(500);
        return b64.toUtf8();
    }
    return decoded;
}

QList<QByteArray> JsonRpcClient::extractJsonFrames(QByteArray& buffer) {
    QList<QByteArray> frames;
    int start = -1;
    int braceDepth = 0;
    bool inString = false;
    char prev = 0;
    for (int i = 0; i < buffer.size(); ++i) {
        char c = buffer[i];
        if (c == '"' && prev != '\\') inString = !inString;
        if (!inString) {
            if (c == '{') {
                if (braceDepth == 0) start = i;
                braceDepth++;
            } else if (c == '}') {
                braceDepth--;
                if (braceDepth == 0 && start != -1) {
                    int len = i - start + 1;
                    frames.append(buffer.mid(start, len));
                    start = -1;
                }
            }
        }
        prev = c;
    }
    if (braceDepth == 0) {
        if (!frames.isEmpty()) {
            int lastEnd = buffer.lastIndexOf('}');
            if (lastEnd != -1) buffer = buffer.mid(lastEnd + 1);
        }
    } else {
        if (start > 0) buffer = buffer.mid(start);
    }
    return frames;
}

bool JsonRpcClient::waitAndReadResponse(int expectId, QJsonObject& outResp, int timeoutMs) {
    QElapsedTimer timer; timer.start();
    while (timer.elapsed() < timeoutMs) {
        if (socket_.bytesAvailable() == 0) {
            int remain = timeoutMs - (int)timer.elapsed();
            if (!socket_.waitForReadyRead(qMin(remain, 1000))) {
                continue;
            }
        }
        QByteArray chunk = socket_.readAll();
        if (!chunk.isEmpty()) recvBuffer_.append(chunk);
        auto frames = extractJsonFrames(recvBuffer_);
        for (const QByteArray& f : frames) {
            QJsonParseError perr;
            QJsonDocument jd = QJsonDocument::fromJson(f, &perr);
            if (perr.error != QJsonParseError::NoError) {
                qWarning() << "[JsonRpcClient] parse frame error:" << perr.errorString();
                continue;
            }
            if (jd.isArray()) {
                for (const QJsonValue& v : jd.array()) {
                    if (!v.isObject()) continue;
                    QJsonObject obj = v.toObject();
                    if (obj.value("id").toInt() == expectId) { outResp = obj; return true; }
                }
            } else if (jd.isObject()) {
                QJsonObject obj = jd.object();
                if (obj.value("id").toInt() == expectId) { outResp = obj; return true; }
            }
        }
    }
    return false;
}

QJsonObject JsonRpcClient::sendRequestBuildAndAwait(const QJsonObject& req, int timeoutMs) {
    QByteArray payload = QJsonDocument(req).toJson(QJsonDocument::Compact);
    int id = req.value("id").toInt();
    if (socket_.write(payload) == -1 || !socket_.waitForBytesWritten(1000)) {
        return QJsonObject{{"error", QJsonObject{{"code", -1},{"message", "write failed"}}}};
    }
    QJsonObject resp;
    if (!waitAndReadResponse(id, resp, timeoutMs)) {
        return QJsonObject{{"error", QJsonObject{{"code", -2},{"message", "timeout"}}}};
    }
    return resp;
}

void JsonRpcClient::writeCall(const QString& method, const QJsonArray& params, const QString& module) {
    call(method, params, module);
}

void JsonRpcClient::writeCall(const QByteArray& requestByte, const int& id) {
    QMutexLocker locker(&mutex_);
    if (socket_.state() != QAbstractSocket::ConnectedState) {
        if (!reconnect()) {
            qWarning() << "write failure: no connnection";
            return;
        }
    }

    QJsonObject resp;
    if (socket_.write(requestByte) == -1 || !socket_.waitForBytesWritten(1000)) {
        qWarning() << "write failure: failed";
        resp = QJsonObject{ {"error", QJsonObject{{"code", -1},{"message", "write failed"}}} };
        return;
    }

    if (!waitAndReadResponse(id, resp, 8000)) {
        qWarning() << "write failure: no response";
        return;
    }
    if (resp.contains("error")) {
        qWarning() << "[JsonRpcClient] Error:" << QJsonDocument(resp["error"].toObject()).toJson(QJsonDocument::Compact);
        return;
    }
    return;
}