Server 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 server:

    example::v1::Example::AsyncService service;
    std::unique_ptr<grpc::Server> server;
    grpc::ServerBuilder builder;
    agrpc::GrpcContext grpc_context{builder.AddCompletionQueue()};
    builder.AddListeningPort("0.0.0.0:50051", grpc::InsecureServerCredentials());
    builder.RegisterService(&service);
    server = builder.BuildAndStart();
    example::ServerShutdown shutdown{*server, grpc_context};
    register_handlers(grpc_context, service);
    grpc_context.run();

Unary rpc

void server_rpc_unary(agrpc::GrpcContext& grpc_context,
                      example::v1::Example::AsyncService& service)
{
    using RPC = agrpc::ServerRPC<&example::v1::Example::AsyncService::RequestUnary>;
    agrpc::register_awaitable_rpc_handler<RPC>(
        grpc_context, service,
        [](RPC& rpc, RPC::Request& request) -> asio::awaitable<void>
        {
            RPC::Response response;
            response.set_integer(request.integer());
            co_await rpc.finish(response, grpc::Status::OK);
 
            // Alternatively finish with an error:
            co_await rpc.finish_with_error(grpc::Status::CANCELLED);
        },
        asio::detached);
}

Client-streaming rpc

void server_rpc_client_streaming(agrpc::GrpcContext& grpc_context,
                                 example::v1::Example::AsyncService& service)
{
    using RPC =
        agrpc::ServerRPC<&example::v1::Example::AsyncService::RequestClientStreaming>;
    agrpc::register_awaitable_rpc_handler<RPC>(
        grpc_context, service,
        [](RPC& rpc) -> asio::awaitable<void>
        {
            RPC::Request request;
            while (co_await rpc.read(request))
            {
                std::cout << "Request: " << request.integer() << std::endl;
            }
            RPC::Response response;
            response.set_integer(42);
            co_await rpc.finish(response, grpc::Status::OK);
 
            // Alternatively finish with an error:
            co_await rpc.finish_with_error(grpc::Status::CANCELLED);
        },
        asio::detached);
}

Server-streaming rpc

void server_rpc_server_streaming(agrpc::GrpcContext& grpc_context,
                                 example::v1::Example::AsyncService& service)
{
    using RPC =
        agrpc::ServerRPC<&example::v1::Example::AsyncService::RequestServerStreaming>;
    agrpc::register_awaitable_rpc_handler<RPC>(
        grpc_context, service,
        [](RPC& rpc, RPC::Request& request) -> asio::awaitable<void>
        {
            RPC::Response response;
            for (int i{}; i != request.integer(); ++i)
            {
                response.set_integer(i);
                if (!co_await rpc.write(response))
                {
                    co_return;
                }
            }
            co_await rpc.finish(grpc::Status::OK);
        },
        asio::detached);
}

Bidirectional-streaming rpc

void server_rpc_bidirectional_streaming(agrpc::GrpcContext& grpc_context,
                                        example::v1::Example::AsyncService& service)
{
    using RPC = agrpc::ServerRPC<
        &example::v1::Example::AsyncService::RequestBidirectionalStreaming>;
    agrpc::register_awaitable_rpc_handler<RPC>(
        grpc_context, service,
        [](RPC& rpc) -> asio::awaitable<void>
        {
            RPC::Request request;
            RPC::Response response;
            while (co_await rpc.read(request))
            {
                response.set_integer(request.integer());
                if (!co_await rpc.write(response))
                {
                    co_return;
                }
            }
            response.set_integer(42);
            co_await rpc.write(response, grpc::WriteOptions{}.set_last_message());
            co_await rpc.finish(grpc::Status::OK);
        },
        asio::detached);
}