std::error_code> {
mContext->TryCancel();
return {};
}
})
co_return std::nullopt;

co_return element;
}

asyncio::task::Task<bool> write(const RequestElement element) { /* same as Writer */ }
asyncio::task::Task<bool> writeDone() { /* same as Writer */ }
asyncio::task::Task<void> done() { /* same as Reader */ }

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

When using a bidirectional stream, reading and writing are two independent tasks that run concurrently:

asyncio::task::Task<void> chat(std::shared_ptr<grpc::ClientContext> context) {
Stream<sample::ChatMessage, sample::ChatMessage> stream{context};
stub->async()->Chat(context.get(), &stream);

stream.AddHold();
stream.StartCall();

co_await all(
// Read task
asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
while (const auto msg = co_await stream.read()) {
fmt::print("Received: {}\n", msg->content());
}
}),
// Write task
asyncio::task::spawn([&]() -> asyncio::task::Task<void> {
sample::ChatMessage msg;
msg.set_content("Hello!");
co_await stream.write(msg);
co_await stream.writeDone();
})
);

stream.RemoveHold();
co_await stream.done();
}

`all()` waits for both the read and write subtasks simultaneously. If either fails, it cancels the other and returns the error. This is the task-tree mechanism of asyncio in action — structured concurrency.

# Wrapping GenericClient

The three streaming wrappers above all follow the same pattern, so it is time to unify them with templates. `GenericClient` provides one `call` overload for each of the four RPC types:

template<typename T>
class GenericClient {
using Stub = T::Stub;
using AsyncStub = class Stub::async;

public:
explicit GenericClient(std::unique_ptr<Stub> stub) : mStub{std::move(stub)} {
}

protected:
// 1. Unary RPC
template<typename Request, typename Response>
asyncio::task::Task<Response>
call(
void (AsyncStub::*method)(grpc::ClientContext *, const Request *, Response *,
std::function<void(grpc::Status)>),
std::shared_ptr<grpc::ClientContext> context,
Request request
) {
Response response;
asyncio::Promise<void, std::string> promise;

std::invoke(
method,
mStub->async(),
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;
}

// 2. Server streaming: push data into a channel via Sender
template<typename Request, typename Element>
asyncio::task::Task<void>
call(