# 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`](/docs/routing) 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`).

```toml
[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`](https://docs.rs/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:

```rust
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`:

```rust
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:

```rust
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:

```rust
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:

```rust
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](/docs/transports/http)):

<Tabs items="['Tokio','Compio']">
  <Tab value="Tokio">
    ```rust
    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(())
    }
    ```
  </Tab>

  <Tab value="Compio">
    ```rust
    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(())
    }
    ```
  </Tab>
</Tabs>

<Callout type="info">
  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.
</Callout>

## 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:

```rust
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.

<Callout type="warn">
  Compressed messages are **not** supported. A request whose compressed-flag byte is
  set is rejected with `Unimplemented`, signalling the client to retry uncompressed.
</Callout>

## Calling it with grpcurl

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

```bash
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](/docs/transports/sse) or
[WebSocket](/docs/transports/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.

Source: [examples/grpc-streaming/src/main.rs](https://github.com/rust-dd/tako/blob/main/examples/grpc-streaming/src/main.rs)

```rust
//! 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(())
}
```

Source: https://tako.rust-dd.com/docs/transports/grpc