EnergySpectrumAnalyer/src/MeasureClient/MeasureClient.cpp

#include "MeasureClient.h"
#include "MeasureServiceProtocol.h"
#include <QByteArray>
#include <QTimer>
#include <QJsonDocument>
#include <QDataStream>
#include <QDebug>

MeasureClient::MeasureClient(QObject *parent)
    : QObject(parent),
    _is_auto_reconnect(true),
    _response_buffer(new QByteArray()),
    _response_data_len(0)
{
    _host = "127.0.0.1";
    _port = 9999;

    _tcp_socket = new QTcpSocket(this);
    _reconnect_timer = new QTimer(this);

    connect(_tcp_socket, &QTcpSocket::connected, this, &MeasureClient::onConnected);
    connect(_tcp_socket, &QTcpSocket::disconnected, this, &MeasureClient::onDisconnected);
    connect(_tcp_socket, &QTcpSocket::readyRead, this, &MeasureClient::onReadyRead);
    connect(_tcp_socket, QOverload<QAbstractSocket::SocketError>::of(&QTcpSocket::error), this, &MeasureClient::onSocketError);
    connect(_reconnect_timer, &QTimer::timeout, this, &MeasureClient::tryReconnect);
    _reconnect_timer->setInterval(300);
}

MeasureClient::~MeasureClient()
{
    if (_tcp_socket->isOpen()) {
        _tcp_socket->close();
    }
}

void MeasureClient::connectToServer(const QString &ip, quint16 port)
{
    QMutexLocker locker(&_mutex);
    _host = ip;
    _port = port;
    _tcp_socket->connectToHost(_host, _port);
}

void MeasureClient::disconnectFromServer()
{
    _is_auto_reconnect = false;
    _reconnect_timer->stop();
    if (_tcp_socket->isOpen()) {
        _tcp_socket->disconnectFromHost();
        _tcp_socket->close();
    }
}

void MeasureClient::startMeasure(const QString &device_guid, const QVariantMap &config)
{
    QString data;
    if (!config.isEmpty()) {
        QJsonDocument json_doc = QJsonDocument::fromVariant(config);
        data = json_doc.toJson(QJsonDocument::Compact);
    }
    QByteArray request_data;
    QDataStream request_stream(&request_data, QIODevice::WriteOnly);
    request_stream << QString("START") << device_guid;
    if (!data.isEmpty()) {
        request_stream << data;
    }
    this->sendRequst(request_data);
}

void MeasureClient::stopMeasure(const QString &device_guid)
{
    QByteArray request_data;
    QDataStream request_stream(&request_data, QIODevice::WriteOnly);
    request_stream << QString("STOP") << device_guid;
    this->sendRequst(request_data);
}

void MeasureClient::setMeasureConfigParams(const QString &device_guid, const QVariantMap &config)
{
    QByteArray request_data;
    QDataStream request_stream(&request_data, QIODevice::WriteOnly);
    request_stream << QString("SET") << device_guid;
    this->sendRequst(request_data);
}

void MeasureClient::clearData(const QString &device_guid)
{
    QByteArray request_data;
    QDataStream request_stream(&request_data, QIODevice::WriteOnly);
    request_stream << QString("CLEAR") << device_guid;
    this->sendRequst(request_data);
}

void MeasureClient::getDeviceList()
{
    QByteArray request_data;
    QDataStream request_stream(&request_data, QIODevice::WriteOnly);
    request_stream << QString("DEVICE");
    this->sendRequst(request_data);
}

void MeasureClient::onConnected()
{
    _reconnect_timer->stop();
    emit runningInfo(QStringLiteral(u"连接到服务器"));
}

void MeasureClient::onDisconnected()
{
    if (_is_auto_reconnect) {
        emit runningInfo(QStringLiteral(u"重连服务器......"));
        _reconnect_timer->start();
    } else {
        emit runningInfo(QStringLiteral(u"断开服务器连接"));
    }
}

void MeasureClient::onReadyRead()
{
    _response_buffer->append(_tcp_socket->readAll());
    while (true) {
        if (_response_data_len == 0) {
            if (_response_buffer->size() < Protocol::HEAD_SIZE) break;
            QByteArray head = _response_buffer->left(Protocol::HEAD_SIZE);
            _response_data_len = Protocol::UnpackDataLen(head);
            _response_buffer->remove(0, Protocol::HEAD_SIZE);
        }
        if (_response_buffer->size() >= _response_data_len) {
            QByteArray response_data = _response_buffer->left(_response_data_len);
            _response_buffer->remove(0, _response_data_len);
            QThread* process_thread = QThread::create(&MeasureClient::processResponse, this, response_data);
            connect(process_thread, &QThread::finished, process_thread, &QThread::deleteLater);
            process_thread->start();
            _response_data_len = 0;
        } else {
            break;
        }
    }
}

void MeasureClient::onSocketError(QAbstractSocket::SocketError error)
{
    emit errorOccurred(QStringLiteral(u"连接异常:%1").arg(_tcp_socket->errorString()));
    if (error == QAbstractSocket::ConnectionRefusedError) {
        _reconnect_timer->start();
    }
}

void MeasureClient::tryReconnect()
{
    _tcp_socket->connectToHost(_host, _port);
}

void MeasureClient::sendRequst(const QByteArray &data)
{
    if (!_tcp_socket->isOpen()) {
        emit errorOccurred(QStringLiteral(u"未连接服务器，请求发送失败"));
        return;
    }
    QByteArray requst_buf = Protocol::PackData(data);
    _tcp_socket->write(requst_buf);
    _tcp_socket->flush();
}

void MeasureClient::processResponse(const QByteArray &response_data)
{
    QDataStream response_stream(response_data);
    QString response_command;
    bool success;
    response_stream >> response_command >> success;
    if (response_command == "GVF") {
        int data_size;
        response_stream >> data_size;
        QByteArray data_buf;
        data_buf.resize(data_size);
        response_stream.readRawData(data_buf.data(), data_size);
        emit gvfData(data_buf);
    } else if (response_command == "START") {
        if (success) {
            QString measure_gvf_data_filename;
            response_stream >> measure_gvf_data_filename;
            emit startMeasureResult(success, measure_gvf_data_filename);
        } else {
            QString message;
            response_stream >> message;
            emit startMeasureResult(success, message);
        }
    } else if (response_command == "DEVICE") {
        QStringList devices;
        QString message;
        if (success) {
            int count;
            response_stream >> count;
            for (int i = 0; i < count; ++i) {
                QString device;
                response_stream >> device;
                devices << device;
            }
        } else {
            response_stream >> message;
        }
        emit getDeviceListResult(success, message, devices);
    } else if (response_command == "STOP") {
        QString message;
        response_stream >> message;
        emit stopMeasureResult(success, message);
    } else if (response_command == "SET") {
        QString message;
        response_stream >> message;
        emit setMeasureConfigParamsResult(success, message);
    } else if (response_command == "CLEAR") {
        QString message;
        response_stream >> message;
        emit clearDataResult(success, message);
    } else {
        QString message;
        response_stream >> message;
        emit errorOccurred(message);
    }
}