C++ - Reddit

consume a server-streaming RPC:

asyncio::task::Task<void> getNumbers(
const sample::GetNumbersRequest &request,
std::shared_ptr<grpc::ClientContext> context
) {
Reader<sample::Number> reader{context};
stub->async()->GetNumbers(context.get(), &request, &reader);

reader.AddHold();
reader.StartCall();

while (true) {
auto number = co_await reader.read();

if (!number)
break; // stream ended

fmt::print("Received: {}\n", number->value());
}

reader.RemoveHold();
co_await reader.done(); // wait for the stream to fully finish and check for errors
}

>`AddHold` and `RemoveHold` are gRPC Reactor lifecycle control mechanisms that prevent the Reactor from being destroyed while we hold it.

# Client Streaming (Writer)

Client-streaming RPC is similar to server-streaming but in the opposite direction. Inherit `grpc::ClientWriteReactor<T>`; after each `StartWrite` completes, `OnWriteDone` is called back:

template<typename T>
class Writer final : public grpc::ClientWriteReactor<T> {
public:
void OnWriteDone(const bool ok) override {
std::exchange(mWritePromise, std::nullopt)->resolve(ok);
}

void OnWritesDoneDone(const bool ok) override {
std::exchange(mWriteDonePromise, std::nullopt)->resolve(ok);
}

asyncio::task::Task<bool> write(const T element) {
asyncio::Promise<bool> promise;
auto future = promise.getFuture();

mWritePromise.emplace(std::move(promise));
grpc::ClientWriteReactor<T>::StartWrite(&element);

co_return co_await asyncio::task::Cancellable{
std::move(future),
[this]() -> std::expected<void, std::error_code> {
mContext->TryCancel();
return {};
}
};
}

asyncio::task::Task<bool> writeDone() {
asyncio::Promise<bool> promise;
auto future = promise.getFuture();

mWriteDonePromise.emplace(std::move(promise));
grpc::ClientWriteReactor<T>::StartWritesDone();

co_return co_await asyncio::task::Cancellable{
std::move(future),
[this]() -> std::expected<void, std::error_code> {
mContext->TryCancel();
return {};
}
};
}

// OnDone and done() are similar to Reader, omitted for brevity

private:
std::shared_ptr<grpc::ClientContext> mContext;
asyncio::Promise<void, std::string> mDonePromise;
std::optional<asyncio::Promise<bool>> mWritePromise;
std::optional<asyncio::Promise<bool>> mWriteDonePromise;
};

`writeDone()` corresponds to gRPC's `StartWritesDone`, which signals to the server that the client has finished writing — equivalent to sending EOF on the stream.

# Client Bidirectional Streaming

Bidirectional streaming is the most complex of the four patterns: the client simultaneously has both read and write capabilities. Fortunately, all that is needed is to merge the `Reader` and `Writer` logic into a single `Stream` class:

template<typename RequestElement, typename ResponseElement>
class Stream final : public grpc::ClientBidiReactor<RequestElement, ResponseElement> {
public:
// OnReadDone, OnWriteDone, OnWritesDoneDone, OnDone are the same as before

asyncio::task::Task<std::optional<ResponseElement>> read() {
ResponseElement element;

asyncio::Promise<bool> promise;
auto future = promise.getFuture();

mReadPromise.emplace(std::move(promise));
grpc::ClientBidiReactor<RequestElement, ResponseElement>::StartRead(&element);

if (!co_await asyncio::task::Cancellable{
std::move(future),
[this]() -> std::expected<void,

6 views12:23