Server streaming: потоковая передача данных от сервера к Qt-клиенту

Введение

Unary RPC из предыдущей статьи подходит для разовых запросов, но для непрерывного потока обновлений (новые показания датчика, поступающие в реальном времени) более естественно подходит server streaming — RPC, при котором сервер может отправить клиенту множество сообщений в ответ на один запрос, без необходимости клиенту повторно запрашивать каждое следующее значение, что концептуально похоже на WebSocket из основного цикла статей про сеть, но с той же строгой типизацией Protobuf.

Концепция

В .proto-файле server streaming обозначается ключевым словом stream перед типом возвращаемого значения метода — сгенерированный серверный код предоставляет объект ServerWriter, через который можно отправить произвольное количество сообщений одно за другим, прежде чем метод вернёт управление (завершив поток). На клиентской стороне вызов такого метода возвращает объект-читатель, через который сообщения принимаются одно за другим в цикле, пока поток не будет завершён сервером.

Пример кода: Server streaming

// Дополнение к device_service.proto — RPC с потоковым ответом сервера
service TelemetryService {
    rpc GetLatestReading(ReadingRequest) returns (SensorReading);
    rpc GetMultipleReadings(DeviceListRequest) returns (DeviceListResponse);

    // stream перед типом ответа означает, что сервер может отправить много сообщений подряд
    rpc StreamReadings(ReadingRequest) returns (stream SensorReading);
}

// server/telemetry_server.cpp — реализация потокового метода на сервере
grpc::Status StreamReadings(grpc::ServerContext *context,
                              const telemetry::ReadingRequest *request,
                              grpc::ServerWriter<telemetry::SensorReading> *writer) override
{
    const std::string &deviceId = request->device_id();

    // Отправляем новое показание раз в секунду, пока клиент не отключится
    // (context->IsCancelled() становится true, если клиент закрыл соединение)
    while (!context->IsCancelled()) {
        telemetry::SensorReading reading;
        reading.set_device_id(deviceId);
        reading.set_temperature(20.0 + (rand() % 100) / 10.0); // имитация показаний для примера
        reading.set_timestamp_unix(static_cast<int64_t>(time(nullptr)));

        if (!writer->Write(reading)) {
            break; // запись не удалась — соединение разорвано
        }

        std::this_thread::sleep_for(std::chrono::seconds(1));
    }

    return grpc::Status::OK;
}

// client/telemetryclient.h — приём потока показаний на стороне клиента
void streamReadingsBlocking(const QString &deviceId)
{
    telemetry::ReadingRequest request;
    request.set_device_id(deviceId.toStdString());

    grpc::ClientContext context;
    std::unique_ptr<grpc::ClientReader<telemetry::SensorReading>> reader(
        m_stub->StreamReadings(&context, request));

    telemetry::SensorReading reading;

    // Read() блокирует до получения следующего сообщения; возвращает false,
    // когда поток завершён сервером или соединение прервано
    while (reader->Read(&reading)) {
        qDebug() << "Получено показание:" << reading.temperature()
                 << "в момент" << reading.timestamp_unix();
    }

    grpc::Status status = reader->Finish();
    if (!status.ok()) {
        qWarning() << "Поток завершился с ошибкой:" << QString::fromStdString(status.error_message());
    }
}

Пояснения к коду

Серверная реализация StreamReadings принимает grpc::ServerWriter<SensorReading> *writer вместо обычного указателя на единственный объект ответа — каждый вызов writer->Write(reading) отправляет клиенту ещё одно сообщение, и метод продолжает выполняться (в примере — в цикле while), пока явно не решит завершиться (возвратом из функции) или пока клиент не отключится, что обнаруживается через context->IsCancelled(). Это принципиально отличается от unary RPC, где метод завершается сразу после заполнения единственного объекта ответа.

На клиенте m_stub->StreamReadings(&context, request) возвращает не готовый ответ, а объект-читатель (ClientReader), и reader->Read(&reading) вызывается в цикле, блокируясь на каждой итерации до получения следующего сообщения от сервера — цикл естественным образом завершается, когда Read() возвращает false (поток закрыт сервером), после чего reader->Finish() возвращает финальный статус всего потокового вызова, который нужно явно проверить отдельно от успешности отдельных полученных сообщений.

Подводные камни

Блокирующий вызов reader->Read() в цикле, выполняемый в основном потоке Qt-приложения, останавливает обработку событий интерфейса на неопределённо долгое время (пока поток активен) — это даже более выраженная версия проблемы из статьи 324, и для server streaming, по самой своей природе предполагающего длительное, непрерывное получение данных, асинхронная интеграция с циклом событий Qt (статья 327) практически обязательна, а не просто желательна.
Бесконечный цикл на сервере (while (!context->IsCancelled())) без какого-либо ограничения по времени или количеству отправленных сообщений — если клиент «забыл» корректно закрыть соединение (например, из-за краша без штатного закрытия), серверный поток выполнения может оставаться активным неопределённо долго, потребляя ресурсы сервера; для production-кода стоит предусмотреть дополнительные ограничения (максимальная длительность потока, явный тайм-аут неактивности) сверх простой проверки отмены контекста.
Отсутствие обработки ситуации, когда writer->Write() возвращает false (что означает разрыв соединения или другую проблему передачи) без явного break — продолжение попыток записи в уже нерабочий поток после такой ошибки бессмысленно и может привести к неопределённому поведению; пример корректно обрабатывает это через break, но забытая проверка результата Write() — частая ошибка в наивной реализации.
Использование std::this_thread::sleep_for() непосредственно внутри обработчика gRPC-метода на сервере блокирует поток, обслуживающий именно этот конкретный вызов — для сервера с ограниченным пулом потоков (а не выделенным потоком на каждое соединение, что характерно для большинства практичных конфигураций gRPC) это снижает максимальное количество одновременно обслуживаемых клиентов; для производственного сервера с большим количеством одновременных потоковых клиентов стоит рассмотреть асинхронный серверный API gRPC вместо простого синхронного с блокирующими задержками внутри обработчика.