gRPC

Serve unary, server-streaming, client-streaming, and bidirectional gRPC over HTTP/2 on ordinary Tako routes, on Tokio or Compio.

Tako serves gRPC by treating each method as a normal HTTP route. There is no separate gRPC server: a gRPC handler is an ordinary async function registered on the same Router as your REST endpoints. All four RPC shapes are supported: unary, server streaming, client streaming, and bidirectional. The helpers live in tako-rs-core behind grpc and work on either runtime. Use an HTTP/2 listener: h2c or TLS on either runtime (TLS on Compio needs compio-tls).

[dependencies]
tako-rs = { version = "2", features = ["grpc", "http2"] }
prost = "0.14"
futures-util = "0.3"

The grpc feature gates the tako::grpc module. Protobuf message types come from prost; in production they are generated by prost-build from a .proto file. futures-util provides the StreamExt combinators used by streaming handlers.

RPC shape	Handler takes	Handler returns
Unary	`GrpcRequest<Req>`	`GrpcResponse<Resp>`
Server streaming	`GrpcRequest<Req>`	`GrpcServerStream<S, Resp>`
Client streaming	`GrpcClientStream<Req>`	`GrpcResponse<Resp>`
Bidirectional	`GrpcBidi<Req, Resp>`	`GrpcServerStream<S, Resp>`

Defining messages

Request and reply types are prost::Message derives. Inline definitions are fine for a quick service; real projects generate them:

use prost::Message;

#[derive(Clone, PartialEq, Message)]
pub struct HelloRequest {
  #[prost(string, tag = "1")]
  pub name: String,
}

#[derive(Clone, PartialEq, Message)]
pub struct HelloReply {
  #[prost(string, tag = "1")]
  pub message: String,
}

#[derive(Clone, PartialEq, Message)]
pub struct Number {
  #[prost(int64, tag = "1")]
  pub value: i64,
}

Unary RPCs

GrpcRequest<T> decodes the gRPC-framed protobuf body and exposes it as req.message. GrpcResponse<U> frames the reply and sets the gRPC headers. Build a success with GrpcResponse::ok(msg) or an error with GrpcResponse::error(code, message) where code is a GrpcStatusCode:

use tako::grpc::{GrpcRequest, GrpcResponse, GrpcStatusCode};

async fn say_hello(req: GrpcRequest<HelloRequest>) -> GrpcResponse<HelloReply> {
  let name = &req.message.name;
  if name.is_empty() {
    return GrpcResponse::error(GrpcStatusCode::InvalidArgument, "name must not be empty");
  }
  GrpcResponse::ok(HelloReply {
    message: format!("Hello, {name}!"),
  })
}

A successful reply sends the message and then grpc-status: 0 in the HTTP/2 trailers. An error is a trailers-only response: HTTP 200 OK with grpc-status and grpc-message in the headers and no body, per the gRPC HTTP/2 mapping.

Server streaming

Return a GrpcServerStream built from any Stream of Result<T, GrpcStatus>. Each Ok item is sent as one message as soon as the stream yields it, and the call ends with grpc-status: 0 when the stream finishes:

use futures_util::StreamExt;
use tako::grpc::{GrpcRequest, GrpcServerStream};
use tako::responder::Responder;

async fn count(req: GrpcRequest<Number>) -> impl Responder {
  let numbers = futures_util::stream::iter(1..=req.message.value)
    .map(|value| Ok(Number { value }));
  GrpcServerStream::new(numbers)
}

The first Err(status) ends the call with that status, and later items are not polled. Use with_metadata(headers) to send initial metadata with the response headers.

Client streaming

GrpcClientStream<T> is a Stream of Result<T, GrpcError> that yields messages while the client is still sending them. The ? operator converts a GrpcError into a GrpcStatus, so a handler that returns Result<_, GrpcStatus> can bail out on a bad message:

use futures_util::StreamExt;
use tako::grpc::{GrpcClientStream, GrpcResponse, GrpcStatus};

async fn sum(mut numbers: GrpcClientStream<Number>) -> Result<GrpcResponse<Number>, GrpcStatus> {
  let mut value = 0;
  while let Some(number) = numbers.next().await {
    value += number?.value;
  }
  Ok(GrpcResponse::ok(Number { value }))
}

The stream ends after the first error. A request that stops in the middle of a message yields GrpcError::InvalidFrame.

Bidirectional streaming

GrpcBidi<Req, Resp> reads Req messages like GrpcClientStream and names the reply type. respond builds the GrpcServerStream from the inbound stream. The reply starts as soon as the handler returns, so both directions are open at the same time and each reply can go out before the client sends its next message:

use futures_util::StreamExt;
use tako::grpc::{GrpcBidi, GrpcStatus};
use tako::responder::Responder;

async fn running_total(bidi: GrpcBidi<Number, Number>) -> impl Responder {
  bidi.respond(|numbers| {
    numbers.scan(0, |total, number| {
      let reply = number
        .map(|number| {
          *total += number.value;
          Number { value: *total }
        })
        .map_err(GrpcStatus::from);
      futures_util::future::ready(Some(reply))
    })
  })
}

When replies do not map one-to-one onto requests, spawn a task that reads the inbound stream, and return a stream over the receiving end of a channel, such as tokio_stream::wrappers::ReceiverStream.

Status and errors

GrpcStatusCode is the full canonical set (Ok, InvalidArgument, NotFound, DeadlineExceeded, Unimplemented, ResourceExhausted, …). A GrpcStatus pairs a code with an optional message:

in a unary handler, return GrpcResponse::error(code, message);
in a stream, yield Err(GrpcStatus::error(code, message)) to end the call after the messages already sent;
to reject a call before any message, return Result<_, GrpcStatus> from the handler. GrpcStatus is itself a responder that sends a trailers-only response.

Status messages are percent-encoded as the spec requires, so non-ASCII text survives the trip.

Attaching the service to the router

A gRPC method maps to POST /<package>.<Service>/<Method>. Register it like any other route. gRPC requires HTTP/2 on the wire, so enable the http2 feature and serve over a transport that negotiates h2: try_spawn_h2c for cleartext behind a proxy, or TLS with ALPN (see HTTP):

use tako::Method;
use tako::router::Router;
use tako::types::BoxError;

#[tokio::main]
async fn main() -> Result<(), BoxError> {
  let mut router = Router::new();
  router.route(Method::POST, "/grpc.Greeter/SayHello", say_hello);

  let listener = tokio::net::TcpListener::bind("127.0.0.1:8080").await?;
  tako::Server::builder()
    .build()
    .try_spawn_h2c(listener, router)?
    .result()
    .await?;
  Ok(())
}

use tako::Method;
use tako::router::Router;
use tako::types::BoxError;

#[compio::main]
async fn main() -> Result<(), BoxError> {
  let mut router = Router::new();
  router.route(Method::POST, "/grpc.Greeter/SayHello", say_hello);

  let listener = compio::net::TcpListener::bind("127.0.0.1:8080").await?;
  tako::CompioServer::builder()
    .build()
    .try_spawn_h2c(listener, router)?
    .result()
    .await?;
  Ok(())
}

The path string is the gRPC contract — it must be the exact /<package>.<Service>/<Method> a client like grpcurl, Envoy, or a generated stub will call. There is no reflection-based auto-registration; you list each method explicitly.

Deadlines and cancellation

An Option<GrpcDeadline> handler argument reads the client's grpc-timeout header. Pass it to with_deadline and the stream ends with DeadlineExceeded once the deadline passes, even if the stream is waiting for its next item:

use std::time::Duration;

use tako::grpc::{GrpcDeadline, GrpcRequest, GrpcServerStream};
use tako::responder::Responder;

async fn slow_count(deadline: Option<GrpcDeadline>, req: GrpcRequest<Number>) -> impl Responder {
  let last = req.message.value;
  let numbers = futures_util::stream::unfold(1, move |value| async move {
    if value > last {
      return None;
    }
    tokio::time::sleep(Duration::from_millis(200)).await;
    Some((Ok(Number { value }), value + 1))
  });
  GrpcServerStream::new(numbers).with_deadline(deadline)
}

In a unary handler, bound the work with deadline.remaining(), for example with tokio::time::timeout. Parsing is overflow-safe: an absurdly large value is treated as "no deadline" rather than panicking. read_grpc_deadline(&mut req) stores the deadline in the request extensions for middleware, and the extractor reuses it.

When a client cancels a call or disconnects, the server drops the reply stream together with everything it owns, so a producer task notices through its closed channel. An inbound stream whose client goes away yields GrpcError::BodyReadError.

Message size limits

Each inbound message is capped at MAX_GRPC_MESSAGE_SIZE (4 MiB), matching the default grpc-go and tonic server limits, for unary and streaming requests alike. A length prefix that advertises more is rejected with ResourceExhausted before the message is buffered, so a hostile client cannot force a large allocation with a few bytes. The extractors read the body message by message and never buffer it whole.

Compressed messages are not supported. A request whose compressed-flag byte is set is rejected with Unimplemented, signalling the client to retry uncompressed.

Calling it with grpcurl

Tako has no server reflection, so pass the .proto file to the client:

grpcurl -plaintext -import-path proto -proto numbers.proto \
  -d '{"value": 3}' 127.0.0.1:50051 numbers.Numbers/Count
grpcurl -plaintext -import-path proto -proto numbers.proto \
  -d '{"value": 1} {"value": 2} {"value": 3}' 127.0.0.1:50051 numbers.Numbers/RunningTotal

Generated tonic clients work too: the streaming calls are tested against a tonic client on both runtimes.

Other helpers

The module also contains a gRPC-Web bridge, a GrpcInterceptor chain, and grpc.health.v1 / reflection scaffolding. These are lower-level building blocks; for browser-facing realtime, prefer SSE or WebSocket.

Examples

examples/grpc-unary: a Greeter and an Echo service registered as POST routes.
examples/grpc-streaming: server, client, and bidirectional streaming with a proto/numbers.proto for grpcurl.

examples/grpc-streaming/src/main.rs

//! Server-streaming, client-streaming, and bidirectional gRPC on Tako.
//!
//! Run the server, then call it with grpcurl from this directory:
//!
//! ```text
//! grpcurl -plaintext -import-path proto -proto numbers.proto \
//!   -d '{"value": 3}' 127.0.0.1:50051 numbers.Numbers/Count
//! grpcurl -plaintext -import-path proto -proto numbers.proto \
//!   -d '{"value": 1} {"value": 2} {"value": 3}' 127.0.0.1:50051 numbers.Numbers/Sum
//! grpcurl -plaintext -import-path proto -proto numbers.proto \
//!   -d '{"value": 1} {"value": 2} {"value": 3}' 127.0.0.1:50051 numbers.Numbers/RunningTotal
//! ```

use std::time::Duration;

use futures_util::StreamExt;
use prost::Message;
use tako::Method;
use tako::grpc::GrpcBidi;
use tako::grpc::GrpcClientStream;
use tako::grpc::GrpcDeadline;
use tako::grpc::GrpcRequest;
use tako::grpc::GrpcResponse;
use tako::grpc::GrpcServerStream;
use tako::grpc::GrpcStatus;
use tako::grpc::GrpcStatusCode;
use tako::responder::Responder;
use tako::router::Router;
use tako::types::BoxError;

/// Generated by `prost-build` from `proto/numbers.proto` in a real project.
#[derive(Clone, PartialEq, Message)]
pub struct Number {
  #[prost(int64, tag = "1")]
  pub value: i64,
}

/// Server streaming: one request, a stream of replies.
async fn count(
  deadline: Option<GrpcDeadline>,
  req: GrpcRequest<Number>,
) -> Result<impl Responder, GrpcStatus> {
  let last = req.message.value;
  if !(1..=100).contains(&last) {
    return Err(GrpcStatus::error(
      GrpcStatusCode::InvalidArgument,
      "value must be between 1 and 100",
    ));
  }
  let numbers = futures_util::stream::unfold(1, move |value| async move {
    if value > last {
      return None;
    }
    tokio::time::sleep(Duration::from_millis(200)).await;
    Some((Ok(Number { value }), value + 1))
  });
  // Ends the call with DEADLINE_EXCEEDED if the client's `grpc-timeout` passes.
  Ok(GrpcServerStream::new(numbers).with_deadline(deadline))
}

/// Client streaming: a stream of requests, one reply.
async fn sum(mut numbers: GrpcClientStream<Number>) -> Result<GrpcResponse<Number>, GrpcStatus> {
  let mut value = 0;
  while let Some(number) = numbers.next().await {
    value += number?.value;
  }
  Ok(GrpcResponse::ok(Number { value }))
}

/// Bidirectional streaming: each reply is sent as soon as its request arrives.
async fn running_total(bidi: GrpcBidi<Number, Number>) -> impl Responder {
  bidi.respond(|numbers| {
    numbers.scan(0, |total, number| {
      let reply = number
        .map(|number| {
          *total += number.value;
          Number { value: *total }
        })
        .map_err(GrpcStatus::from);
      futures_util::future::ready(Some(reply))
    })
  })
}

#[tokio::main]
async fn main() -> Result<(), BoxError> {
  let mut router = Router::new();
  router.route(Method::POST, "/numbers.Numbers/Count", count);
  router.route(Method::POST, "/numbers.Numbers/Sum", sum);
  router.route(Method::POST, "/numbers.Numbers/RunningTotal", running_total);

  let listener = tokio::net::TcpListener::bind("127.0.0.1:50051").await?;
  println!("gRPC (h2c) on http://127.0.0.1:50051");

  tako::Server::builder()
    .build()
    .try_spawn_h2c(listener, router)?
    .result()
    .await?;
  Ok(())
}

On this page