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,
