eUi/rpcModule/rpc_finsTcp_static/rpc_finsTcp.cpp

#include "rpc_finsTcp.h"
#include "../RpcExeProcessThread.h"

#include <QDebug>
#include <QElapsedTimer>
#include <QMutexLocker>

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

void destory(rpcModuleBase* ptr)
{
    delete ptr;
    qDebug() << "rpc_finsTcp destory success!";
}

rpc_finsTcp::rpc_finsTcp() = default;

QByteArray rpc_finsTcp::makeExecuteRequest(const QByteArray& command, quint32& id)
{
    id = idCounter.fetch_add(1);
    return RpcFrameCodec::build(
        id,
        RpcFrameCodec::CmdExecute,
        RpcFrameCodec::requestPayload("finstcp_instance_0", QString::fromUtf8(command)));
}

QByteArray rpc_finsTcp::executeReadCommand(QByteArray& input)
{
    RpcClient* client = nullptr;
    {
        QMutexLocker locker(&mutex);
        client = readClients.value(input);
        if (!client) {
            client = new RpcClient("127.0.0.1", 9000);
            if (client->_connectionStatus != Connected) {
                delete client;
                return QByteArray();
            }
            if (!client->initialize("finstcp_instance_0", m_IP, m_port)) {
                qWarning() << "[rpc_finsTcp] read client initialization failed"
                           << m_IP << m_port;
                delete client;
                return QByteArray();
            }
            readClients.insert(input, client);
        }
    }
    return client->execute("finstcp_instance_0", input);
}

bool rpc_finsTcp::executeWriteCommand(QByteArray& input)
{
    if (input == "null") {
        return true;
    }
    if (input.startsWith("init_")) {
        return parseInitCommand(input);
    }

    const QList<QByteArray> parts = input.split('_');
    if (parts.size() < 4 || parts.at(0) == "read") {
        qWarning() << "[rpc_finsTcp] asynchronous write rejected: invalid command" << input;
        return false;
    }
    quint32 id = 0;
    const QByteArray request = makeExecuteRequest(input, id);
    return enqueueWrite(request, id, input);
}

rpc_finsTcp::~rpc_finsTcp()
{
    if (writeClient) {
        writeClient->stopWorker();
        delete writeClient;
        writeClient = nullptr;
    }
    qDeleteAll(readClients);
    readClients.clear();
    stopRpcServer();
    qDebug() << "~rpc_finsTcp success!";
}

bool rpc_finsTcp::readData(std::vector<cv::Mat>& vec_mat, QByteArray& data)
{
    Q_UNUSED(vec_mat);
    RpcClient* client = nullptr;
    {
        QMutexLocker locker(&mutex);
        client = readClients.value(data);
        if (!client) {
            client = new RpcClient("127.0.0.1", 9000);
            if (client->_connectionStatus != Connected) {
                delete client;
                return false;
            }
            client->initialize("finstcp_instance_0", m_IP, m_port);
            readClients.insert(data, client);
        }
    }

    const QByteArray response = client->execute("finstcp_instance_0", data);
    if (response.isEmpty()) {
        return false;
    }
    data = response;
    return true;
}

bool rpc_finsTcp::writeData(std::vector<cv::Mat>& vec_mat, QByteArray& data)
{
    Q_UNUSED(vec_mat);
    const QList<QByteArray> parts = data.split('_');
    if (parts.size() < 3) {
        qWarning() << "[rpc_finsTcp] asynchronous write rejected: invalid command" << data;
        return false;
    }

    if (parts.at(0) == "init") {
        m_IP = parts.at(1);
        m_port = parts.at(2).toInt();
    } else if (data.contains("read")) {
        return true;
    } else {
        if (data.contains("null")) {
            return true;
        }
        if (!data.contains("Heart")) {
            qDebug() << data;
        }
        quint32 id = 0;
        const QByteArray request = makeExecuteRequest(data, id);
        if (!enqueueWrite(request, id, data)) {
            return false;
        }
    }
    return true;
}

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

    if (!startRpcServer()) {
        return false;
    }

    writeClient = new RpcClientWorker();
    if (!writeClient->startAndWait()) {
        qWarning() << "[rpc_finsTcp] asynchronous writer thread failed to connect";
        delete writeClient;
        writeClient = nullptr;
        stopRpcServer();
        return false;
    }
    if (!sendInitCommand()) {
        writeClient->stopWorker();
        delete writeClient;
        writeClient = nullptr;
        stopRpcServer();
        return false;
    }
    return true;
}

bool rpc_finsTcp::startRpcServer()
{
    if (!processThread) {
        processThread = new RpcExeProcessThread();
    }
    if (processThread->startExe("post_rpc_server_1_2_0.exe")) {
        return true;
    }
    delete processThread;
    processThread = nullptr;
    return false;
}

void rpc_finsTcp::stopRpcServer()
{
    if (!processThread) {
        return;
    }
    processThread->stopExe();
    delete processThread;
    processThread = nullptr;
}

bool rpc_finsTcp::parseInitCommand(const QByteArray& input)
{
    const QList<QByteArray> parts = input.split('_');
    if (parts.size() < 3 || parts.at(0) != "init") {
        qCritical() << input << "expected init_ip_port";
        return false;
    }

    bool portOk = false;
    const int port = parts.at(2).toInt(&portOk);
    if (parts.at(1).isEmpty() || !portOk || port <= 0 || port > 65535) {
        qCritical() << input << "invalid init parameters";
        return false;
    }
    m_IP = QString::fromUtf8(parts.at(1));
    m_port = port;
    return true;
}

bool rpc_finsTcp::sendInitCommand()
{
    const quint32 id = idCounter.fetch_add(1);
    const QByteArray request = RpcFrameCodec::build(
        id,
        RpcFrameCodec::CmdInit,
        RpcFrameCodec::requestPayload(
            "finstcp_instance_0", m_IP + "\n" + QString::number(m_port)));
    return enqueueWrite(request, id,
                        ("init_" + m_IP + "_" + QString::number(m_port)).toUtf8());
}

bool rpc_finsTcp::run()
{
    return true;
}

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

bool RpcClient::reconnect()
{
    if (socket_.state() == QAbstractSocket::ConnectedState) {
        return true;
    }
    socket_.abort();
    codec_.clear();
    socket_.connectToHost(host_, port_);
    return socket_.waitForConnected(3000);
}

bool RpcClient::initialize(const QString& module, const QString& ip, int port)
{
    QMutexLocker locker(&mutex_);
    if (!reconnect()) {
        return false;
    }
    const quint32 id = idCounter_++;
    const QByteArray payload = RpcFrameCodec::requestPayload(
        module, ip + "\n" + QString::number(port));
    const QByteArray frame = RpcFrameCodec::build(id, RpcFrameCodec::CmdInit, payload);
    RpcFrameCodec::Frame response;
    return sendFrameAndAwait(frame, id, response) &&
           response.cmd == RpcFrameCodec::RespOk;
}

bool RpcClient::isConnected(const QString& module)
{
    QMutexLocker locker(&mutex_);
    if (!reconnect()) {
        return false;
    }
    const quint32 id = idCounter_++;
    const QByteArray frame = RpcFrameCodec::build(
        id, RpcFrameCodec::CmdIsConnected,
        RpcFrameCodec::requestPayload(module));
    RpcFrameCodec::Frame response;
    if (!sendFrameAndAwait(frame, id, response) || response.cmd != RpcFrameCodec::RespOk) {
        return false;
    }
    return !response.payload.isEmpty() && response.payload.at(0) != '\0';
}

QByteArray RpcClient::execute(const QString& module, const QByteArray& command)
{
    QMutexLocker locker(&mutex_);
    if (!reconnect()) {
        return QByteArray();
    }
    const quint32 id = idCounter_++;
    const QByteArray frame = RpcFrameCodec::build(
        id, RpcFrameCodec::CmdExecute,
        RpcFrameCodec::requestPayload(module, QString::fromUtf8(command)));
    RpcFrameCodec::Frame response;
    if (!sendFrameAndAwait(frame, id, response) || response.cmd != RpcFrameCodec::RespOk) {
        qWarning() << "[RpcClient] execute request failed";
        return QByteArray();
    }
    return response.payload;
}

bool RpcClient::sendFrame(const QByteArray& request, quint32 id)
{
    QMutexLocker locker(&mutex_);
    if (!reconnect()) {
        qWarning() << "[RpcClient] asynchronous write failed: RPC server is not connected";
        return false;
    }
    RpcFrameCodec::Frame response;
    if (!sendFrameAndAwait(request, id, response)) {
        qWarning() << "[RpcClient] asynchronous write failed: response timeout or socket error"
                   << "requestId=" << id;
        return false;
    }
    if (response.cmd != RpcFrameCodec::RespOk) {
        qWarning() << "[RpcClient] asynchronous write rejected by server"
                   << "requestId=" << id
                   << "response=" << response.payload;
        return false;
    }
    return true;
}

bool RpcClient::waitAndReadResponse(quint32 expectId,
                                    RpcFrameCodec::Frame& response,
                                    int timeoutMs)
{
    QElapsedTimer timer;
    timer.start();
    while (timer.elapsed() < timeoutMs) {
        if (socket_.bytesAvailable() == 0) {
            const int remain = timeoutMs - static_cast<int>(timer.elapsed());
            if (!socket_.waitForReadyRead(qMin(remain, 1000))) {
                continue;
            }
        }
        const QByteArray chunk = socket_.readAll();
        if (!chunk.isEmpty()) {
            codec_.append(chunk);
        }
        const QList<RpcFrameCodec::Frame> frames = codec_.takeFrames();
        for (const RpcFrameCodec::Frame& frame : frames) {
            if (frame.id == expectId) {
                response = frame;
                return true;
            }
        }
    }
    return false;
}

bool RpcClient::sendFrameAndAwait(const QByteArray& frame, quint32 id,
                                  RpcFrameCodec::Frame& response, int timeoutMs)
{
    if (socket_.write(frame) == -1 || !socket_.waitForBytesWritten(1000)) {
        qWarning() << "[RpcClient] frame write failed";
        return false;
    }
    return waitAndReadResponse(id, response, timeoutMs);
}

RpcClientWorker::~RpcClientWorker()
{
    stopWorker();
}

bool RpcClientWorker::startAndWait(int timeoutMs)
{
    if (!isRunning()) {
        start();
    }
    QMutexLocker locker(&stateMutex_);
    while (!initialized_) {
        if (!initializedCondition_.wait(&stateMutex_, timeoutMs)) {
            qWarning() << "writer init timeout";
            return false;
        }
    }
    return connection_ == Connected;
}

bool RpcClientWorker::enqueue(const QByteArray& request, quint32 id,
                              const QByteArray& command)
{
    QMutexLocker locker(&queueMutex_);
    if (stopping_ || !isReady()) {
        qWarning() << "write rejected: writer not ready"
                   << "requestId=" << id;
        return false;
    }
    queue_.enqueue({request, command, id});
    queueCondition_.wakeOne();
    return true;
}

bool RpcClientWorker::isReady() const
{
    QMutexLocker locker(&stateMutex_);
    return initialized_ && connection_ == Connected;
}

void RpcClientWorker::stopWorker()
{
    {
        QMutexLocker locker(&queueMutex_);
        stopping_ = true;
        queue_.clear();
        queueCondition_.wakeAll();
    }
    if (isRunning() && QThread::currentThread() != this) {
        if (!wait(10000)) {
            qWarning() << "writer stop timeout";
        }
    }
}

void RpcClientWorker::run()
{
    RpcClient client;
    {
        QMutexLocker locker(&stateMutex_);
        connection_ = client._connectionStatus;
        initialized_ = true;
        initializedCondition_.wakeAll();
    }

    for (;;) {
        PendingWrite pending;
        {
            QMutexLocker locker(&queueMutex_);
            while (queue_.isEmpty() && !stopping_) {
                queueCondition_.wait(&queueMutex_);
            }
            if (stopping_) {
                break;
            }
            pending = queue_.dequeue();
        }
        if (!client.sendFrame(pending.request, pending.id)) {
            qWarning() << "write failed" << pending.command;
        }
        else  if (!pending.command.contains("Heart")) {
            qDebug() << pending.command;
        }
    }
}

bool rpc_finsTcp::enqueueWrite(const QByteArray& request, quint32 id,
                               const QByteArray& command)
{
    if (!writeClient) {
        qWarning() << "[rpc_finsTcp] asynchronous write rejected: writer thread is not initialized"
                   << "requestId=" << id;
        return false;
    }
    return writeClient->enqueue(request, id, command);
}