Client rpc cheat sheet

Table of Contents

• Unary rpc
• Client-streaming rpc
• Server-streaming rpc
• Bidirectional-streaming rpc

The code below is based on example.proto.

A single-threaded gRPC client:

    agrpc::GrpcContext grpc_context;
    example::v1::Example::Stub stub(grpc::CreateChannel("localhost:50051", grpc::InsecureChannelCredentials()));
    asio::co_spawn(
        grpc_context,
        [&]() -> asio::awaitable<void>
        {
            // ...
            co_return;
        },
        asio::detached);
    grpc_context.run();

Unary rpc

asio::awaitable<void> client_rpc_unary(agrpc::GrpcContext& grpc_context,
                                       example::v1::Example::Stub& stub)
{
    using RPC = asio::use_awaitable_t<>::as_default_on_t<
        agrpc::ClientRPC<&example::v1::Example::Stub::PrepareAsyncUnary>>;
    grpc::ClientContext client_context;
    client_context.set_deadline(std::chrono::system_clock::now() +
                                std::chrono::seconds(5));
    RPC::Request request;
    RPC::Response response;
    const grpc::Status status =
        co_await RPC::request(grpc_context, stub, client_context, request, response);
    if (!status.ok())
    {
        std::cerr << "Rpc failed: " << status.error_message();
        co_return;
    }
    std::cout << "Response: " << response.integer();
}

Client-streaming rpc

asio::awaitable<void> client_rpc_client_streaming(agrpc::GrpcContext& grpc_context,
                                                  example::v1::Example::Stub& stub)
{
    using RPC = asio::use_awaitable_t<>::as_default_on_t<
        agrpc::ClientRPC<&example::v1::Example::Stub::PrepareAsyncClientStreaming>>;
 
    RPC rpc{grpc_context};
    rpc.context().set_deadline(std::chrono::system_clock::now() +
                               std::chrono::seconds(5));
 
    RPC::Response response;
    if (!co_await rpc.start(stub, response))
    {
        const grpc::Status status = co_await rpc.finish();
        std::cerr << "Rpc failed: " << status.error_message();
        co_return;
    }
 
    RPC::Request request;
    request.set_integer(1);
    while (co_await rpc.write(request) && request.integer() < 42)
    {
        request.set_integer(request.integer() + 1);
    }
 
    const grpc::Status status = co_await rpc.finish();
    if (!status.ok())
    {
        std::cerr << "Rpc failed: " << status.error_message();
        co_return;
    }
 
    std::cout << "Response: " << response.integer();
}

Server-streaming rpc

asio::awaitable<void> client_rpc_server_streaming(agrpc::GrpcContext& grpc_context,
                                                  example::v1::Example::Stub& stub)
{
    using RPC = asio::use_awaitable_t<>::as_default_on_t<
        agrpc::ClientRPC<&example::v1::Example::Stub::PrepareAsyncServerStreaming>>;
 
    RPC rpc{grpc_context};
    rpc.context().set_deadline(std::chrono::system_clock::now() +
                               std::chrono::seconds(5));
 
    RPC::Request request;
    request.set_integer(42);
    if (!co_await rpc.start(stub, request))
    {
        const grpc::Status status = co_await rpc.finish();
        std::cerr << "Rpc failed: " << status.error_message();
        co_return;
    }
 
    RPC::Response response;
    while (co_await rpc.read(response))
    {
        std::cout << "Response: " << response.integer() << '\n';
    }
 
    const grpc::Status status = co_await rpc.finish();
    if (!status.ok())
    {
        std::cerr << "Rpc failed: " << status.error_message();
        co_return;
    }
}

Bidirectional-streaming rpc

asio::awaitable<void> client_rpc_bidirectional_streaming(agrpc::GrpcContext& grpc_context,
                                                         example::v1::Example::Stub& stub)
{
    using RPC = asio::use_awaitable_t<>::as_default_on_t<agrpc::ClientRPC<
        &example::v1::Example::Stub::PrepareAsyncBidirectionalStreaming>>;
 
    RPC rpc{grpc_context};
    rpc.context().set_deadline(std::chrono::system_clock::now() +
                               std::chrono::seconds(5));
 
    if (!co_await rpc.start(stub))
    {
        const grpc::Status status = co_await rpc.finish();
        std::cerr << "Rpc failed: " << status.error_message();
        co_return;
    }
 
    RPC::Request request;
    request.set_integer(42);
 
    bool write_ok{true};
    RPC::Response response;
    while (co_await rpc.read(response) && write_ok)
    {
        request.set_integer(response.integer() + 1);
        write_ok = co_await rpc.write(request);
    }
 
    const grpc::Status status = co_await rpc.finish();
    if (!status.ok())
    {
        std::cerr << "Rpc failed: " << status.error_message();
        co_return;
    }
}