GrabBag/App/ParkingSpaceGuide/ParkingSpaceGuideView/RemoteGuideClient.cpp

#include "RemoteGuideClient.h"

#include <QByteArray>
#include <QFutureWatcher>
#include <QHostAddress>
#include <QJsonDocument>
#include <QJsonParseError>
#include <QMetaObject>
#include <QNetworkAddressEntry>
#include <QNetworkInterface>
#include <QPointer>
#include <QSet>
#include <QStringList>
#include <QTimer>
#include <QUdpSocket>
#include <QtConcurrent/QtConcurrentRun>

#include "IVrZeroMQClient.h"
#include "IVrZeroMQPubSub.h"
#include "VrLog.h"

namespace {

const char kProtocol[] = "ParkingSpaceGuideRemote";
const char kService[] = "ParkingSpaceGuide";

bool IsValidPort(int port)
{
    return port > 0 && port <= 65535;
}

QString ResponseMessage(const QJsonObject& response,
                        const QString& fallback)
{
    const QString message = response.value("message").toString().trimmed();
    return message.isEmpty() ? fallback : message;
}

} // namespace

RemoteGuideClient::RemoteGuideClient(int discoveryPort, QObject* parent)
    : QObject(parent)
    , m_discoveryPort(IsValidPort(discoveryPort) ? discoveryPort : 5555)
    , m_discoveryTimer(new QTimer(this))
    , m_heartbeatTimer(new QTimer(this))
    , m_requestWatcher(new QFutureWatcher<ControlReply>(this))
{
    m_discoveryTimer->setInterval(1000);
    m_heartbeatTimer->setInterval(3000);

    connect(m_discoveryTimer, &QTimer::timeout,
            this, &RemoteGuideClient::SendDiscovery);
    connect(m_heartbeatTimer, &QTimer::timeout, this, [this]() {
        if (m_connected && !m_requestInFlight) {
            SendControlCommand(QStringLiteral("ping"));
        }
    });
    connect(m_requestWatcher, &QFutureWatcher<ControlReply>::finished,
            this, &RemoteGuideClient::HandleControlFinished);

    EnsureDiscoverySocket();
    LOG_INFO("ParkingSpaceGuideView discovery started localPort=%u targetPort=%d\n",
             m_discoverySocket
                 ? static_cast<unsigned int>(m_discoverySocket->localPort())
                 : 0U,
             m_discoveryPort);
    m_discoveryTimer->start();
    QTimer::singleShot(0, this, &RemoteGuideClient::SendDiscovery);
}

RemoteGuideClient::~RemoteGuideClient()
{
    ++m_generation;
    m_discoveryTimer->stop();
    m_heartbeatTimer->stop();
    disconnect(m_requestWatcher, nullptr, this, nullptr);
    if (m_requestWatcher->isRunning()) {
        m_requestWatcher->waitForFinished();
    }
    if (m_subscriber) {
        m_subscriber->UnInit();
        delete m_subscriber;
        m_subscriber = nullptr;
    }
}

bool RemoteGuideClient::IsConnected() const
{
    return m_connected;
}

bool RemoteGuideClient::IsGuidanceRunning() const
{
    return m_guidanceRunning;
}

QString RemoteGuideClient::ServerAddress() const
{
    return m_serverAddress;
}

void RemoteGuideClient::SelectModel(const QString& modelType)
{
    QJsonObject arguments;
    arguments["modelType"] = modelType.trimmed().toUpper();
    SendControlCommand(QStringLiteral("select_model"), arguments);
}

void RemoteGuideClient::StartGuidance(const QString& modelType)
{
    QJsonObject arguments;
    arguments["modelType"] = modelType.trimmed().toUpper();
    SendControlCommand(QStringLiteral("start"), arguments);
}

void RemoteGuideClient::StopGuidance()
{
    SendControlCommand(QStringLiteral("stop"));
}

void RemoteGuideClient::RequestStatus()
{
    SendControlCommand(QStringLiteral("get_status"));
}

void RemoteGuideClient::EnsureDiscoverySocket()
{
    if (!m_discoverySocket) {
        m_discoverySocket = new QUdpSocket(this);
        connect(m_discoverySocket, &QUdpSocket::readyRead,
                this, &RemoteGuideClient::ReadDiscoveryDatagrams);
    }
    if (m_discoverySocket->state() == QAbstractSocket::UnconnectedState) {
        const bool bound = m_discoverySocket->bind(
            QHostAddress::AnyIPv4,
            0,
            QUdpSocket::ShareAddress | QUdpSocket::ReuseAddressHint);
        if (!bound) {
            const QByteArray errorText = m_discoverySocket->errorString()
                                             .toUtf8();
            LOG_ERROR("ParkingSpaceGuideView discovery bind failed error=%s\n",
                      errorText.constData());
        }
    }
}

void RemoteGuideClient::SendDiscovery()
{
    if (m_connected) {
        return;
    }

    EnsureDiscoverySocket();
    QJsonObject request;
    request["cmd"] = QStringLiteral("discover");
    request["service"] = QString::fromLatin1(kService);
    request["protocol"] = QString::fromLatin1(kProtocol);
    const QByteArray payload = QJsonDocument(request)
                                   .toJson(QJsonDocument::Compact);

    QSet<QString> broadcastAddresses;
    broadcastAddresses.insert(QHostAddress(QHostAddress::LocalHost).toString());
    broadcastAddresses.insert(QHostAddress(QHostAddress::Broadcast).toString());
    const QList<QNetworkInterface> interfaces =
        QNetworkInterface::allInterfaces();
    for (const QNetworkInterface& networkInterface : interfaces) {
        const QNetworkInterface::InterfaceFlags flags = networkInterface.flags();
        if (!flags.testFlag(QNetworkInterface::IsUp) ||
            !flags.testFlag(QNetworkInterface::IsRunning) ||
            flags.testFlag(QNetworkInterface::IsLoopBack)) {
            continue;
        }
        for (const QNetworkAddressEntry& entry :
             networkInterface.addressEntries()) {
            if (entry.ip().protocol() != QAbstractSocket::IPv4Protocol ||
                entry.broadcast().isNull()) {
                continue;
            }
            broadcastAddresses.insert(entry.broadcast().toString());
        }
    }

    ++m_discoveryAttempt;
    bool sent = false;
    int sentCount = 0;
    for (const QString& address : broadcastAddresses) {
        const qint64 sentBytes = m_discoverySocket->writeDatagram(
            payload,
            QHostAddress(address),
            static_cast<quint16>(m_discoveryPort));
        if (sentBytes >= 0) {
            sent = true;
            ++sentCount;
        } else {
            const QByteArray addressText = address.toUtf8();
            const QByteArray errorText = m_discoverySocket->errorString()
                                             .toUtf8();
            LOG_WARNING("ParkingSpaceGuideView discovery send failed target=%s:%d error=%s\n",
                        addressText.constData(),
                        m_discoveryPort,
                        errorText.constData());
        }
    }
    const QByteArray targetsText = QStringList(broadcastAddresses.values())
                                       .join(QStringLiteral(","))
                                       .toUtf8();
    LOG_INFO("ParkingSpaceGuideView discovery attempt=%d localPort=%u targetPort=%d sent=%d targets=%s\n",
             m_discoveryAttempt,
             static_cast<unsigned int>(m_discoverySocket->localPort()),
             m_discoveryPort,
             sentCount,
             targetsText.constData());
    SetConnectionMessage(sent
        ? QStringLiteral("正在搜索停机引导设备……")
        : QStringLiteral("UDP搜索发送失败，将继续重试：%1")
              .arg(m_discoverySocket->errorString()));
}

void RemoteGuideClient::ReadDiscoveryDatagrams()
{
    while (m_discoverySocket && m_discoverySocket->hasPendingDatagrams()) {
        QByteArray payload;
        payload.resize(static_cast<int>(
            m_discoverySocket->pendingDatagramSize()));
        QHostAddress sender;
        quint16 senderPort = 0;
        const qint64 bytes = m_discoverySocket->readDatagram(
            payload.data(), payload.size(), &sender, &senderPort);
        if (bytes <= 0) {
            continue;
        }
        payload.resize(static_cast<int>(bytes));
        const QByteArray senderText = sender.toString().toUtf8();
        LOG_INFO("ParkingSpaceGuideView discovery datagram peer=%s:%u bytes=%lld\n",
                 senderText.constData(),
                 static_cast<unsigned int>(senderPort),
                 static_cast<long long>(bytes));

        QJsonParseError error;
        const QJsonDocument document = QJsonDocument::fromJson(payload, &error);
        if (error.error != QJsonParseError::NoError ||
            !document.isObject()) {
            const QByteArray errorText = error.errorString().toUtf8();
            LOG_WARNING("ParkingSpaceGuideView invalid discovery JSON peer=%s:%u error=%s\n",
                        senderText.constData(),
                        static_cast<unsigned int>(senderPort),
                        errorText.constData());
            continue;
        }
        const QJsonObject object = document.object();
        if (!object.value("ok").toBool() ||
            object.value("cmd").toString() != QStringLiteral("discover") ||
            object.value("protocol").toString() !=
                QString::fromLatin1(kProtocol) ||
            object.value("service").toString() !=
                QString::fromLatin1(kService)) {
            LOG_WARNING("ParkingSpaceGuideView ignored discovery response peer=%s:%u protocol/service mismatch\n",
                        senderText.constData(),
                        static_cast<unsigned int>(senderPort));
            continue;
        }
        if (!m_connected) {
            LOG_INFO("ParkingSpaceGuideView discovered service peer=%s control=%d publish=%d\n",
                     senderText.constData(),
                     object.value("controlPort").toInt(),
                     object.value("publishPort").toInt());
            ConnectToServer(sender.toString(), object);
        }
    }
}

bool RemoteGuideClient::ConnectToServer(const QString& address,
                                        const QJsonObject& serverInfo)
{
    const int controlPort = serverInfo.value("controlPort").toInt();
    const int publishPort = serverInfo.value("publishPort").toInt();
    const QString topic = serverInfo.value("topic").toString().trimmed();
    if (address.isEmpty() || !IsValidPort(controlPort) ||
        !IsValidPort(publishPort) || topic.isEmpty()) {
        const QByteArray addressText = address.toUtf8();
        LOG_ERROR("ParkingSpaceGuideView invalid service info peer=%s control=%d publish=%d topicEmpty=%d\n",
                  addressText.constData(),
                  controlPort,
                  publishPort,
                  topic.isEmpty() ? 1 : 0);
        return false;
    }

    IVrZeroMQSubscriber* subscriber = nullptr;
    if (!IVrZeroMQSubscriber::CreateObject(&subscriber) || !subscriber) {
        SetConnectionMessage(QStringLiteral("创建状态订阅器失败，将继续搜索"));
        return false;
    }

    const QByteArray expectedTopic = topic.toUtf8();
    const int connectionGeneration = m_generation + 1;
    QPointer<RemoteGuideClient> self(this);
    const FunSubRecv callback =
        [self, expectedTopic, connectionGeneration](const char* topicData,
                                                    size_t topicLength,
                                                    const char* data,
                                                    size_t dataLength) {
            if (!self || !topicData || (!data && dataLength > 0)) {
                return;
            }
            const QByteArray receivedTopic(
                topicData, static_cast<int>(topicLength));
            if (receivedTopic != expectedTopic) {
                return;
            }
            const QByteArray payload(data, static_cast<int>(dataLength));
            QMetaObject::invokeMethod(
                self,
                [self, payload, connectionGeneration]() {
                    if (self && self->m_connected &&
                        self->m_generation == connectionGeneration) {
                        self->HandleStatusPayload(payload);
                    }
                },
                Qt::QueuedConnection);
        };

    const QByteArray addressUtf8 = address.toUtf8();
    const QByteArray topicUtf8 = topic.toUtf8();
    if (subscriber->Init(addressUtf8.constData(),
                         publishPort,
                         topicUtf8.constData(),
                         callback) != 0) {
        subscriber->UnInit();
        delete subscriber;
        SetConnectionMessage(QStringLiteral("连接状态发布端口失败，将继续搜索"));
        return false;
    }

    if (m_subscriber) {
        m_subscriber->UnInit();
        delete m_subscriber;
    }
    m_subscriber = subscriber;
    m_serverAddress = address;
    m_deviceName = serverInfo.value("deviceName")
                       .toString(QStringLiteral("ParkingSpaceGuideApp"));
    m_topic = topic;
    m_controlPort = controlPort;
    m_publishPort = publishPort;
    m_guidanceRunning = serverInfo.value("guidanceRunning").toBool();
    m_connected = true;
    m_generation = connectionGeneration;
    m_discoveryTimer->stop();
    m_heartbeatTimer->start();
    m_connectionMessage = QStringLiteral("已连接");
    const QByteArray serverText = m_serverAddress.toUtf8();
    const QByteArray topicText = m_topic.toUtf8();
    LOG_INFO("ParkingSpaceGuideView connected server=%s control=%d publish=%d topic=%s\n",
             serverText.constData(),
             m_controlPort,
             m_publishPort,
             topicText.constData());
    emit ConnectionStateChanged(true,
                                m_connectionMessage,
                                m_deviceName,
                                m_serverAddress);
    RequestStatus();
    return true;
}

void RemoteGuideClient::DisconnectServer(const QString& reason)
{
    ++m_generation;
    m_connected = false;
    m_guidanceRunning = false;
    m_requestInFlight = false;
    m_heartbeatTimer->stop();
    if (m_subscriber) {
        m_subscriber->UnInit();
        delete m_subscriber;
        m_subscriber = nullptr;
    }
    m_serverAddress.clear();
    m_deviceName.clear();
    m_topic.clear();
    m_controlPort = 0;
    m_publishPort = 0;
    m_connectionMessage = reason;
    emit ConnectionStateChanged(false,
                                reason,
                                QString(),
                                QString());
    if (!m_discoveryTimer->isActive()) {
        m_discoveryTimer->start();
    }
    QTimer::singleShot(0, this, &RemoteGuideClient::SendDiscovery);
}

void RemoteGuideClient::SendControlCommand(
    const QString& command,
    const QJsonObject& arguments)
{
    if (!m_connected) {
        emit CommandFinished(command,
                             false,
                             QStringLiteral("尚未连接停机引导设备"),
                             QJsonObject());
        return;
    }
    if (m_requestInFlight) {
        if (command != QStringLiteral("ping")) {
            QTimer::singleShot(100, this,
                [this, command, arguments]() {
                    SendControlCommand(command, arguments);
                });
        }
        return;
    }

    QJsonObject request = arguments;
    request["cmd"] = command;
    request["protocol"] = QString::fromLatin1(kProtocol);
    const QByteArray payload = QJsonDocument(request)
                                   .toJson(QJsonDocument::Compact);
    const QByteArray address = m_serverAddress.toUtf8();
    const int port = m_controlPort;
    const int generation = m_generation;
    m_requestInFlight = true;

    m_requestWatcher->setFuture(QtConcurrent::run(
        [address, port, payload, command, generation]() {
            ControlReply reply;
            reply.command = command;
            reply.generation = generation;

            IVrZeroMQClient* client = nullptr;
            if (!IVrZeroMQClient::CreateObject(&client) || !client) {
                reply.errorMessage = QStringLiteral("创建控制客户端失败");
                return reply;
            }
            const int initResult = client->Init(address.constData(), port);
            if (initResult != 0) {
                reply.errorMessage = QStringLiteral("连接控制端口失败：%1")
                                         .arg(initResult);
                client->UnInit();
                delete client;
                return reply;
            }

            char* responseData = nullptr;
            size_t responseLength = 0;
            const int sendResult = client->SendAndWaitBack(
                payload.constData(),
                payload.size(),
                &responseData,
                responseLength);
            QByteArray response;
            if (sendResult == 0 && responseData) {
                response = QByteArray(responseData,
                                      static_cast<int>(responseLength));
            }
            client->UnInit();
            delete client;
            if (sendResult != 0) {
                reply.errorMessage = QStringLiteral("控制请求超时或失败：%1")
                                         .arg(sendResult);
                return reply;
            }

            QJsonParseError parseError;
            const QJsonDocument document = QJsonDocument::fromJson(
                response, &parseError);
            if (parseError.error != QJsonParseError::NoError ||
                !document.isObject()) {
                reply.errorMessage = QStringLiteral("服务端返回了无效JSON");
                return reply;
            }
            reply.response = document.object();
            if (reply.response.value("protocol").toString() !=
                QString::fromLatin1(kProtocol)) {
                reply.errorMessage = QStringLiteral("服务端协议不匹配");
                return reply;
            }
            reply.transportOk = true;
            return reply;
        }));
}

void RemoteGuideClient::HandleControlFinished()
{
    const ControlReply reply = m_requestWatcher->result();
    if (reply.generation != m_generation) {
        return;
    }
    m_requestInFlight = false;
    if (!reply.transportOk) {
        if (reply.command != QStringLiteral("ping")) {
            emit CommandFinished(reply.command,
                                 false,
                                 reply.errorMessage,
                                 QJsonObject());
        }
        DisconnectServer(QStringLiteral("通信中断：%1，正在重新搜索设备")
                             .arg(reply.errorMessage));
        return;
    }

    if (reply.response.contains("guidanceRunning")) {
        const bool running =
            reply.response.value("guidanceRunning").toBool();
        if (running != m_guidanceRunning) {
            m_guidanceRunning = running;
            emit GuidanceRunningChanged(running);
        }
    }
    const QJsonObject status = reply.response.value("status").toObject();
    if (!status.isEmpty()) {
        emit StatusReceived(status);
    }

    const bool ok = reply.response.value("ok").toBool();
    if (reply.command != QStringLiteral("ping")) {
        emit CommandFinished(
            reply.command,
            ok,
            ResponseMessage(reply.response,
                            ok ? QStringLiteral("操作成功")
                               : QStringLiteral("服务端拒绝了操作")),
            reply.response);
    }
    if (ok && (reply.command == QStringLiteral("select_model") ||
               reply.command == QStringLiteral("start") ||
               reply.command == QStringLiteral("stop"))) {
        QTimer::singleShot(0, this, &RemoteGuideClient::RequestStatus);
    }
}

void RemoteGuideClient::HandleStatusPayload(const QByteArray& payload)
{
    QJsonParseError error;
    const QJsonDocument document = QJsonDocument::fromJson(payload, &error);
    if (error.error != QJsonParseError::NoError || !document.isObject()) {
        return;
    }
    const QJsonObject status = document.object();
    const QString protocol = status.value("protocol").toString();
    if (!protocol.isEmpty() &&
        protocol != QStringLiteral("ParkingSpaceGuideResult")) {
        return;
    }
    emit StatusReceived(status);
}

void RemoteGuideClient::SetConnectionMessage(const QString& message)
{
    if (m_connected || message == m_connectionMessage) {
        return;
    }
    m_connectionMessage = message;
    const QByteArray messageText = message.toUtf8();
    if (message.contains(QStringLiteral("失败")) ||
        message.contains(QStringLiteral("中断"))) {
        LOG_WARNING("ParkingSpaceGuideView connection state: %s\n",
                    messageText.constData());
    } else {
        LOG_INFO("ParkingSpaceGuideView connection state: %s\n",
                 messageText.constData());
    }
    emit ConnectionStateChanged(false,
                                message,
                                QString(),
                                QString());
}