callback we pass in when the RPC completes. Bridging it with `Promise` is natural:

asyncio::task::Task<sample::EchoResponse> echo(
const sample::EchoRequest &request,
std::shared_ptr<grpc::ClientContext> context
) {
sample::EchoResponse response;
asyncio::Promise<void, std::string> promise;

stub->async()->Echo(
context.get(), &request, &response,
[&](const grpc::Status &status) {
if (!status.ok()) {
promise.reject(status.error_message());
return;
}

promise.resolve();
}
);

if (const auto result = co_await asyncio::task::Cancellable{
promise.getFuture(),
[&]() -> std::expected<void, std::error_code> {
context->TryCancel();
return {};
}
}; !result)
throw co_await asyncio::error::StacktraceError<std::runtime_error>::make(result.error());

co_return response;
}

The core logic is just a few lines: construct a `Promise`, decide `resolve` or `reject` based on `Status` inside the callback, then `co_await` the `Promise`'s `Future`.

Cancellation support is woven in naturally — wrap the `Future` with `Cancellable`, call `context->TryCancel()` on cancellation. From the caller's perspective, this function is indistinguishable from a normal synchronous function, yet it never blocks the event loop.

# Client Streaming RPCs

Streaming RPCs are more complex than Unary, because data is transmitted one piece at a time and each read or write is an independent async operation.

# Server Streaming (Reader)

For a server-streaming RPC, the client needs to inherit `grpc::ClientReadReactor<T>`. Each call to `StartRead` causes gRPC to invoke `OnReadDone` when data is ready.

template<typename T>
class Reader final : public grpc::ClientReadReactor<T> {
public:
explicit Reader(std::shared_ptr<grpc::ClientContext> context) : mContext{std::move(context)} {
}

void OnDone(const grpc::Status &status) override {
if (!status.ok()) {
mDonePromise.reject(status.error_message());
return;
}

mDonePromise.resolve();
}

void OnReadDone(const bool ok) override {
// Resolve the per-read Promise when each read completes
std::exchange(mReadPromise, std::nullopt)->resolve(ok);
}

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

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

mReadPromise.emplace(std::move(promise));
grpc::ClientReadReactor<T>::StartRead(&element);

// ok == false means the stream has ended
if (!co_await asyncio::task::Cancellable{
std::move(future),
[this]() -> std::expected<void, std::error_code> {
mContext->TryCancel();
return {};
}
})
co_return std::nullopt;

co_return element;
}

asyncio::task::Task<void> done() {
if (const auto result = co_await mDonePromise.getFuture(); !result)
throw co_await asyncio::error::StacktraceError<std::runtime_error>::make(result.error());
}

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

A few details worth noting:

* `mReadPromise` is wrapped in `std::optional` because each call to `read()` constructs a new `Promise`.
* `std::exchange` is used in `OnReadDone` to take ownership of the `Promise`, signaling that a single read has completed.
* `done()` waits for the entire stream to finish; it holds a separate `mDonePromise` distinct from the per-read promise.

Using `Reader` to