PowerMaster/source/measurementDataUtils.cpp

#include "measurementDataUtils.h"
#include "logger.h"
#include <QTimer>
#include <QRandomGenerator>

MeasurementData::MeasurementData(const QString&id)
    : m_id(id)
{}

int MeasurementData::insertData(const QVector<MeasurementDataPoint>& newData)
{
    int insertedCount = 0;

    for(const auto& point : newData)
    {
        if (point.timestamp > m_latestTimestamp && !m_timestamps.contains(point.timestamp))
        {

            m_timestamps.insert(point.timestamp);
            m_data.insert(point.timestamp, point.value);
            insertedCount++;

            m_latestTimestamp = point.timestamp;
        }
    }

    return insertedCount;
}

QVector<MeasurementDataPoint> MeasurementData::getLastestData(int count)
{
    QVector<MeasurementDataPoint> result;

    auto it = m_data.constEnd();
    while(it != m_data.constBegin() && result.size() < count)
    {
        --it;
        result.append(MeasurementDataPoint(it.key(), it.value()));
    }

    return result;
}

QVector<MeasurementDataPoint> MeasurementData::getDataInRange(qint64 startTime, qint64 endTime)
{
    QVector<MeasurementDataPoint> result;

    auto end = m_data.upperBound(endTime);
    for(auto it = m_data.lowerBound(startTime); it != end; ++it)
        result.append(MeasurementDataPoint(it.key(), it.value()));

    return result;
}

void MeasurementData::cleanupOldData(qint64 cutoffTime)
{
    auto it = m_data.begin();
    while(it != m_data.end() && it.key() < cutoffTime)
    {
        m_timestamps.remove(it.key());
        it = m_data.erase(it);
    }
}

WebSocketClient::WebSocketClient(QObject* parent)
    : QObject(parent)
    , m_webSocket(nullptr)
    , m_connected(false)
    , m_connectionStatus(ConnectionStatus::Disconnected)
    , m_reconnectTimer(new QTimer(this))
    , m_reconnectInterval(5000) //默认5秒重连
    , m_reconnectAttempts(0)
    , m_maxReconnectAttempts(10) //默认最多重连10次
{
    setupWebSocket();

    m_reconnectTimer->setSingleShot(true);
    connect(m_reconnectTimer, &QTimer::timeout, this, &WebSocketClient::onReconnectTimeout);
}

WebSocketClient::~WebSocketClient()
{
    disconnectFromServer();
    cleanupWebSocket();
}

bool WebSocketClient::connectToServer(const QUrl& url)
{
    if(!url.isValid())
    {
        QString errorMsg = QString("Invalid server URL: %1").arg(url.toString());
        LOG_ERROR("WebSocket", errorMsg);
        return false;
    }

    //if(m_webSocket && (m_webSocket->state() == QAbstractSocket::ConnectedState || m_webSocket->state() == QAbstractSocket::ConnectingState))
    if(m_connectionStatus == ConnectionStatus::Connecting || m_connectionStatus == ConnectionStatus::Connected)
        return true;

    m_serverUrl = url;
    m_connectionStatus = ConnectionStatus::Connecting;

    if(!m_webSocket)
        setupWebSocket();
    m_webSocket->open(url);
    return true;
}

void WebSocketClient::disconnectFromServer()
{
    if(m_webSocket)
    {
        m_webSocket->close();
        m_connected = false;
        m_connectionStatus = ConnectionStatus::Disconnected;

        resetReconnect();
    }
}

void WebSocketClient::setReconnectInterval(int intervalMs)
{
    if(intervalMs > 0 && intervalMs != m_reconnectInterval)
        m_reconnectInterval = intervalMs;
}

void WebSocketClient::setMaxReconnectAttempts(int maxAttempts)
{
    if(maxAttempts >=0 && maxAttempts != m_maxReconnectAttempts)
        m_maxReconnectAttempts = maxAttempts;
}

void WebSocketClient::setupWebSocket()
{
    //cleanupWebSocket();

    m_webSocket = new QWebSocket();
    connect(m_webSocket, &QWebSocket::connected, this, &WebSocketClient::onConnected);
    connect(m_webSocket, &QWebSocket::disconnected, this, &WebSocketClient::onDisconnected);
    connect(m_webSocket, &QWebSocket::errorOccurred, this, &WebSocketClient::onError);
    connect(m_webSocket, &QWebSocket::textMessageReceived, this, &WebSocketClient::onTextMessageReceived);
}

void WebSocketClient::cleanupWebSocket()
{
    if(m_webSocket)
    {
        m_webSocket->disconnected();
        m_webSocket->deleteLater();
        m_webSocket = nullptr;
    }
}

void WebSocketClient::scheduleReconnect()
{
    if(m_connectionStatus == ConnectionStatus::Connected || m_connectionStatus == ConnectionStatus::Connecting
        || m_connectionStatus == ConnectionStatus::Reconncting)
        return;

    if(m_reconnectAttempts < m_maxReconnectAttempts)
    {
        int delay = m_reconnectInterval * (1 /*<< m_reconnectAttempts*/);// 指数退避,<<n相当于乘以2的n次方
        // 添加随机抖动避免多个客户端同时重连
        /*int jitter = QRandomGenerator::global()->bounded(0, 1000);
        delay += jitter;
        // 限制最大延迟为5分钟
        delay = qMin(delay, 300000);*/

        m_reconnectTimer->start(delay);
        m_reconnectAttempts++;
        m_connectionStatus = ConnectionStatus::Reconncting;
    }
}

void WebSocketClient::resetReconnect()
{
    if(m_reconnectTimer->isActive())
        m_reconnectTimer->stop();
    m_reconnectAttempts = 0;
}

///槽函数
void WebSocketClient::onConnected()
{
    m_connected = true;
    m_connectionStatus = ConnectionStatus::Connected;
    resetReconnect();
}

void WebSocketClient::onDisconnected()
{
    m_connected = false;
    m_connectionStatus = ConnectionStatus::Disconnected;
    scheduleReconnect();
}

void WebSocketClient::onError(QAbstractSocket::SocketError error)
{
    m_connected = false;
    m_connectionStatus = ConnectionStatus::Error;
    QString errorMsg = m_webSocket ? m_webSocket->errorString() : "Unknown WebSocket error";
    LOG_ERROR("WebSocket", errorMsg);
    scheduleReconnect();
}

void WebSocketClient::onTextMessageReceived(const QString& message)
{
    emit dataReceived(message);
}

void WebSocketClient::onReconnectTimeout()
{
    if(!m_webSocket)
        setupWebSocket();
    m_webSocket->open(m_serverUrl);
}