Client streaming и bidirectional streaming в Qt-приложении

Введение

Помимо потока от сервера к клиенту (статья 325), gRPC поддерживает обратный случай — client streaming (клиент отправляет серверу множество сообщений, получая один финальный ответ) и bidirectional streaming (оба участника отправляют сообщения друг другу независимо и одновременно). В статье разберём оба этих вида RPC на примере отправки данных от устройства и полнодуплексного канала команд.

Концепция

Client streaming в .proto-файле обозначается stream перед типом запроса — сервер получает сообщения через ServerReader, читая их в цикле, и отправляет единственный финальный ответ после того, как клиент завершит отправку (вызовом WritesDone() на стороне клиента). Bidirectional streaming, обозначаемый stream и для запроса, и для ответа одновременно, даёт обоим участникам полную свободу — отправлять и принимать сообщения в любом порядке и в любой момент, независимо друг от друга, что требует более тщательного проектирования протокола взаимодействия по сравнению с более простыми, однонаправленными видами RPC.

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

// Дополнение к device_service.proto — client streaming и bidirectional streaming

message UploadSummary {
    int32 readings_received = 1;
    double average_temperature = 2;
}

message ControlMessage {
    string command = 1;
}

message ControlAck {
    string command = 1;
    bool success = 2;
}

service TelemetryService {
    // ... предыдущие методы ...

    // Client streaming: клиент отправляет много показаний, получает одну итоговую сводку
    rpc UploadReadings(stream SensorReading) returns (UploadSummary);

    // Bidirectional streaming: команды и подтверждения идут в обе стороны независимо
    rpc ControlChannel(stream ControlMessage) returns (stream ControlAck);
}

// server/telemetry_server.cpp — client streaming: приём множества показаний, одна итоговая сводка
grpc::Status UploadReadings(grpc::ServerContext *context,
                              grpc::ServerReader<telemetry::SensorReading> *reader,
                              telemetry::UploadSummary *summary) override
{
    telemetry::SensorReading reading;
    int count = 0;
    double temperatureSum = 0.0;

    // Read() в цикле — аналогично клиентскому чтению потока из статьи 325,
    // но здесь именно СЕРВЕР читает поток, присылаемый клиентом
    while (reader->Read(&reading)) {
        count++;
        temperatureSum += reading.temperature();
    }

    summary->set_readings_received(count);
    summary->set_average_temperature(count > 0 ? temperatureSum / count : 0.0);

    return grpc::Status::OK;
}

// server/telemetry_server.cpp — bidirectional streaming: обработка команд с подтверждениями
grpc::Status ControlChannel(grpc::ServerContext *context,
                              grpc::ServerReaderWriter<telemetry::ControlAck,
                                                        telemetry::ControlMessage> *stream) override
{
    telemetry::ControlMessage message;

    // Сервер читает входящие команды и сразу же, не дожидаясь завершения всего потока,
    // отправляет подтверждение каждой — это и есть ключевое отличие от client streaming
    while (stream->Read(&message)) {
        telemetry::ControlAck ack;
        ack.set_command(message.command());
        ack.set_success(!message.command().empty());

        stream->Write(ack);
    }

    return grpc::Status::OK;
}

// client/telemetryclient.h — client streaming: отправка нескольких показаний с клиента
telemetry::UploadSummary uploadReadingsBatch(const QVector<SensorReadingData> &readings)
{
    telemetry::UploadSummary summary;
    grpc::ClientContext context;

    std::unique_ptr<grpc::ClientWriter<telemetry::SensorReading>> writer(
        m_stub->UploadReadings(&context, &summary));

    for (const auto &readingData : readings) {
        telemetry::SensorReading reading;
        reading.set_device_id(readingData.deviceId.toStdString());
        reading.set_temperature(readingData.temperature);

        if (!writer->Write(reading)) {
            break; // сервер закрыл поток раньше — дальнейшая отправка бессмысленна
        }
    }

    writer->WritesDone(); // явный сигнал серверу, что отправка завершена — без этого сервер
                           // продолжит ждать ещё сообщений, не переходя к формированию ответа
    grpc::Status status = writer->Finish();

    return summary;
}

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

UploadReadings на сервере принимает ServerReader, а не ServerWriter (в отличие от статьи 325) — роли обратные: теперь сервер читает входящий поток от клиента, накапливая статистику (count, temperatureSum), и только после того, как reader->Read() вернёт false (клиент завершил отправку), заполняет единственный итоговый объект summary и возвращает управление. На клиентской стороне writer->WritesDone() — обязательный, легко забываемый вызов, явно сообщающий серверу, что больше сообщений не будет, и без него reader->Read() на стороне сервера продолжал бы блокироваться, ожидая гипотетических дальнейших сообщений, даже если клиент уже логически закончил отправку всех данных.

ControlChannel показывает наиболее общий, наиболее гибкий вид RPC — ServerReaderWriter, объединяющий возможности и чтения, и записи в одном объекте, что позволяет серверу немедленно отвечать на каждое полученное сообщение (отправляя ControlAck сразу после каждого ControlMessage), не дожидаясь, пока клиент закончит отправку всех своих сообщений — именно это полнодуплексное, не привязанное к строгому порядку «сначала все запросы, потом весь ответ» взаимодействие отличает bidirectional streaming от двух предыдущих, более ограниченных видов RPC.

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

Забытый вызов writer->WritesDone() на клиентской стороне client streaming — без него серверный reader->Read() будет неопределённо долго ждать продолжения потока, который клиент логически уже закончил отправлять, что выглядит как «зависшее» взаимодействие без явной ошибки с обеих сторон, пока не сработает тайм-аут (если он явно настроен, статья 328) или соединение не будет разорвано каким-либо внешним образом.
Проектирование bidirectional streaming без явного протокола соответствия запросов и ответов, если порядок сообщений важен, а сама природа bidirectional streaming не гарантирует, что i-й ответ соответствует именно i-му запросу (в общем случае сервер может обрабатывать и отвечать в любом порядке, особенно при параллельной, многопоточной обработке на стороне сервера) — для случаев, где важно точное сопоставление, протокол должен явно включать идентификаторы запроса в каждом сообщении (аналогично подводному камню о гонке heartbeat-сообщений в статье 289), позволяющие клиенту правильно сопоставить полученный ответ с исходным запросом.
Использование client streaming или bidirectional streaming там, где было бы достаточно простого unary-вызова с массивом (например, repeated SensorReading как одно сообщение в обычном unary RPC вместо потока отдельных сообщений) — потоковые RPC добавляют сложность реализации (явное управление чтением/записью, ошибками на каждом шаге) и оправданы прежде всего когда данных действительно много, они поступают постепенно во времени, или общий объём неизвестен заранее — для небольшого, заранее известного по объёму набора данных простой unary-вызов с повторяющимся полем часто проще в реализации и достаточен.
Отсутствие обработки частичных сбоев в середине длительного потокового взаимодействия (например, разрыв соединения после отправки половины показаний в uploadReadingsBatch) — клиентский код должен явно решить, что делать в этом случае: повторить отправку всех данных с начала, попытаться продолжить с места разрыва (что требует дополнительного протокола отслеживания прогресса), или просто сообщить пользователю о неполной отправке — без явного решения этого вопроса частичный сбой может остаться незамеченным или обработанным непредсказуемым образом.