03. 实时通信与文件通道

void NetworkManager::sendJson(const QJsonObject& json,
                              const QHostAddress& ip,
                              quint16 port)
{
    if (!udpSocket) {
        emit sigLogMessage("⚠️ UDP socket 未初始化，无法发送。");
        return;
    }

    if (ip.isNull() || port == 0) {
        emit sigLogMessage("⚠️ UDP 发送目标无效，已取消发送。");
        return;
    }

    QJsonDocument doc(json);
    QByteArray data = doc.toJson(QJsonDocument::Compact);

    if (data.isEmpty()) {
        emit sigLogMessage("⚠️ UDP 发送内容为空，已取消发送。");
        return;
    }

    if (data.size() > MAX_SAFE_UDP_PACKET_BYTES) {
        emit sigLogMessage(QString("⚠️ UDP 包过大: %1 bytes，可能发生分片丢包。cmd=%2")
                               .arg(data.size())
                               .arg(json.value("cmd").toString()));
    }

    qint64 sent = udpSocket->writeDatagram(data, ip, port);

    if (sent < 0) {
        emit sigLogMessage(QString("⚠️ UDP 发送失败: %1").arg(udpSocket->errorString()));
        return;
    }

    if (sent != data.size()) {
        emit sigLogMessage(QString("⚠️ UDP 发送长度异常: sent=%1 total=%2")
                               .arg(sent)
                               .arg(data.size()));
    }
}

void NetworkManager::processPendingDatagrams()
{
    if (!udpSocket) {
        return;
    }

    int processed = 0;

    while (udpSocket->hasPendingDatagrams()) {
        if (processed >= MAX_DATAGRAMS_PER_TICK) {

            QTimer::singleShot(0, this, &NetworkManager::processPendingDatagrams);
            break;
        }

        processed++;

        QNetworkDatagram datagram = udpSocket->receiveDatagram();
        QByteArray data = datagram.data();

        if (data.isEmpty()) {
            continue;
        }

        if (data.size() > MAX_SAFE_UDP_PACKET_BYTES) {
            emit sigLogMessage(QString("⚠️ 收到过大的 UDP 包: %1 bytes, from=%2:%3")
                                   .arg(data.size())
                                   .arg(datagram.senderAddress().toString())
                                   .arg(datagram.senderPort()));

        }

        QJsonParseError error;
        QJsonDocument doc = QJsonDocument::fromJson(data, &error);

        if (error.error != QJsonParseError::NoError || !doc.isObject()) {
            badDatagramCount++;

            if (badDatagramCount <= 5 || badDatagramCount % 50 == 0) {
                emit sigLogMessage(QString("解析收到的 UDP JSON 失败: %1, from=%2:%3, bad_count=%4")
                                       .arg(error.errorString())
                                       .arg(datagram.senderAddress().toString())
                                       .arg(datagram.senderPort())
                                       .arg(badDatagramCount));
            }

            continue;
        }

        QJsonObject obj = doc.object();

        emit sigJsonReceived(obj);
    }
}

bool NetworkServer::initUdpServer(int port)
{
    if (port <= 0 || port > 65535) {
        std::cerr << "❌ [UDP 服务器] 非法端口: " << port << "\n";
        return false;
    }

    server_fd = socket(AF_INET, SOCK_DGRAM, 0);

    if (server_fd < 0) {
        std::cerr << "❌ [UDP 服务器] 创建 Socket 失败: " << strerror(errno) << "\n";
        return false;
    }

    int reuse = 1;
    if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) < 0) {
        std::cerr << "⚠️ [UDP 服务器] 设置 SO_REUSEADDR 失败: " << strerror(errno) << "\n";
    }

    int flags = fcntl(server_fd, F_GETFL, 0);

    if (flags < 0) {
        std::cerr << "❌ [UDP 服务器] 获取 Socket flags 失败: " << strerror(errno) << "\n";
        close(server_fd);
        server_fd = -1;
        return false;
    }

    if (fcntl(server_fd, F_SETFL, flags | O_NONBLOCK) < 0) {
        std::cerr << "❌ [UDP 服务器] 设置非阻塞失败: " << strerror(errno) << "\n";
        close(server_fd);
        server_fd = -1;
        return false;
    }

    sockaddr_in server_addr{};
    server_addr.sin_family = AF_INET;
    server_addr.sin_addr.s_addr = INADDR_ANY;
    server_addr.sin_port = htons(static_cast<uint16_t>(port));

    if (bind(server_fd, reinterpret_cast<sockaddr*>(&server_addr), sizeof(server_addr)) < 0) {
        std::cerr << "❌ [UDP 服务器] 绑定端口 " << port
                  << " 失败: " << strerror(errno) << "\n";
        close(server_fd);
        server_fd = -1;
        return false;
    }

    epoll_fd = epoll_create1(0);

    if (epoll_fd < 0) {
        std::cerr << "❌ [UDP 服务器] 创建 epoll 失败: " << strerror(errno) << "\n";
        close(server_fd);
        server_fd = -1;
        return false;
    }

    epoll_event event{};
    event.events = EPOLLIN | EPOLLET;
    event.data.fd = server_fd;

    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_fd, &event) < 0) {
        std::cerr << "❌ [UDP 服务器] epoll_ctl 添加监听失败: "
                  << strerror(errno) << "\n";
        close(server_fd);
        close(epoll_fd);
        server_fd = -1;
        epoll_fd = -1;
        return false;
    }

    std::cout << "✅ [UDP 服务器] 已绑定端口 " << port << "，非阻塞 epoll ET 模式\n";
    return true;
}

void TcpFileUploader::startUpload() {
    if (!file->open(QIODevice::ReadOnly)) {
        emit uploadFinished(false, "无法读取本地文件！");
        return;
    }
    payloadSize = file->size();

    qDebug() << "🔗 正在建立 TCP 数据通道连接:" << m_serverIp << ":" << m_serverPort;
    tcpSocket->connectToHost(m_serverIp, m_serverPort);
}

void TcpFileDownloader::startDownload() {
    qDebug() << "🔗 正在连接服务器准备下载...";
    tcpSocket->connectToHost(m_serverIp, m_serverPort);
}

void TcpFileServer::acceptLoop()
{
    while (m_running.load()) {
        sockaddr_in client_addr{};
        socklen_t addr_len = sizeof(client_addr);

        int client_fd = accept(
            m_server_fd,
            reinterpret_cast<sockaddr*>(&client_addr),
            &addr_len
        );

        if (client_fd < 0) {
            if (!m_running.load()) {
                break;
            }

            if (errno == EINTR) {
                continue;
            }

            std::cerr << "⚠️ [TCP 文件服务器] accept 失败: "
                      << strerror(errno) << "\n";
            continue;
        }

        if (m_active_clients.load() >= MAX_ACTIVE_CLIENTS) {
            json resp;
            resp["cmd"] = "error";
            resp["msg"] = "文件服务器繁忙，请稍后重试";
            sendJsonLine(client_fd, resp);
            close(client_fd);
            continue;
        }

        setSocketTimeout(client_fd);

        bool queued = m_worker_pool.tryEnqueue([this, client_fd]() {
            this->handleClient(client_fd);
        });

        if (!queued) {
            json resp;
            resp["cmd"] = "error";
            resp["msg"] = "文件服务器繁忙，请稍后重试";
            sendJsonLine(client_fd, resp);
            close(client_fd);

            std::cerr << "⚠️ [TCP 文件服务器] 线程池队列已满，拒绝新的文件连接。"
                      << " pending=" << m_worker_pool.pendingTasks()
                      << " active=" << m_worker_pool.activeTasks()
                      << " rejected=" << m_worker_pool.rejectedTasks()
                      << "\n";
        }
    }
}