gRPC Streaming, Status Codes, Deadlines, and Interceptors

A gRPC call can send a stream of messages in either direction, and every call has a status, metadata, and an optional deadline. This lesson builds an Orders service with all four call types, status codes, metadata, an authentication interceptor, and a client that uses deadlines. It builds on gRPC basics.

Term	Meaning
Server streaming	One request, many responses
Client streaming	Many requests, one response
Bidirectional streaming	Both sides send messages independently
Metadata	Key-value pairs sent as HTTP/2 headers
Deadline	The time by which a call must finish
Interceptor	A function that inspects a request before the handler runs and can reject it

Project setup

protoc must be installed, as in the basics lesson.

Cargo.toml

TOML

[package]
name = "orders"
version = "0.1.0"
edition = "2021"

[dependencies]
tonic = "0.13"
prost = "0.13"
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "sync"] }
tokio-stream = "0.1"

[build-dependencies]
tonic-build = "0.13"

build.rs

Rust

fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_build::compile_protos("proto/orders.proto")?;
    Ok(())
}

Example 1: The four call types in a schema

The stream keyword on the request, the response, or both selects the call type.

proto/orders.proto

PROTOBUF

syntax = "proto3";

package storefront.orders.v1;

service Orders {
  // Unary
  rpc GetOrder(GetOrderRequest) returns (Order);
  // Server streaming: one request, many responses
  rpc WatchOrder(WatchOrderRequest) returns (stream OrderEvent);
  // Client streaming: many requests, one response
  rpc ImportOrders(stream Order) returns (ImportSummary);
  // Bidirectional streaming
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message GetOrderRequest {
  int64 order_id = 1;
}

message Order {
  int64 id = 1;
  string status = 2;
  int64 total_cents = 3;
}

message WatchOrderRequest {
  int64 order_id = 1;
}

message OrderEvent {
  int64 order_id = 1;
  string status = 2;
}

message ImportSummary {
  int32 imported = 1;
}

message ChatMessage {
  string text = 1;
}

Each message on a stream is length-prefixed, so the receiver reads whole messages one at a time.
A sender ends its side by half-closing. The server ends a call by returning a status.
HTTP/2 flow control applies per stream. A slow reader stops the sender once the window is full, which gives backpressure (the fast side waits for the slow side) with no extra code.

The Rust side is split into one module per call type. service.rs implements the generated trait and delegates each method. A server-streaming method returns a boxed stream, and the trait names it as an associated type. The crate compiles once every file from Examples 1 to 6 exists.

src/lib.rs

Rust

use std::pin::Pin;

use tokio_stream::Stream;
use tonic::Status;

pub mod pb {
    tonic::include_proto!("storefront.orders.v1");
}

pub mod auth;
pub mod chat;
pub mod import;
pub mod service;
pub mod unary;
pub mod watch;

pub type BoxStream<T> = Pin<Box<dyn Stream<Item = Result<T, Status>> + Send>>;

src/service.rs

Rust

use tonic::{Request, Response, Status, Streaming};

use crate::pb::{
    orders_server::Orders, ChatMessage, GetOrderRequest, ImportSummary, Order, OrderEvent,
    WatchOrderRequest,
};
use crate::{chat, import, unary, watch, BoxStream};

#[derive(Default)]
pub struct OrdersService;

#[tonic::async_trait]
impl Orders for OrdersService {
    type WatchOrderStream = BoxStream<OrderEvent>;
    type ChatStream = BoxStream<ChatMessage>;

    async fn get_order(
        &self,
        request: Request<GetOrderRequest>,
    ) -> Result<Response<Order>, Status> {
        unary::get_order(request).await
    }

    async fn watch_order(
        &self,
        request: Request<WatchOrderRequest>,
    ) -> Result<Response<Self::WatchOrderStream>, Status> {
        watch::watch_order(request).await
    }

    async fn import_orders(
        &self,
        request: Request<Streaming<Order>>,
    ) -> Result<Response<ImportSummary>, Status> {
        import::import_orders(request).await
    }

    async fn chat(
        &self,
        request: Request<Streaming<ChatMessage>>,
    ) -> Result<Response<Self::ChatStream>, Status> {
        chat::chat(request).await
    }
}

Example 2: Server streaming

The handler returns a stream of Result<OrderEvent, Status>. A spawned task writes events into a bounded Tokio channel, and the receiver, wrapped as a stream, becomes the response.

src/watch.rs

Rust

use tokio_stream::wrappers::ReceiverStream;
use tonic::{Request, Response, Status};

use crate::pb::{OrderEvent, WatchOrderRequest};
use crate::BoxStream;

pub async fn watch_order(
    request: Request<WatchOrderRequest>,
) -> Result<Response<BoxStream<OrderEvent>>, Status> {
    let order_id = request.into_inner().order_id;
    if order_id <= 0 {
        return Err(Status::invalid_argument("order id must be positive"));
    }

    let (tx, rx) = tokio::sync::mpsc::channel(4);
    tokio::spawn(async move {
        for status in ["placed", "paid", "shipped"] {
            let event = OrderEvent {
                order_id,
                status: status.to_owned(),
            };
            if tx.send(Ok(event)).await.is_err() {
                // The client disconnected or cancelled: stop producing.
                break;
            }
        }
    });

    Ok(Response::new(
        Box::pin(ReceiverStream::new(rx)) as BoxStream<OrderEvent>
    ))
}

The channel holds 4 messages, so the producer waits when the client reads slowly.
tx.send fails when the receiver is dropped, which happens on cancel or disconnect. Checking the error is how a spawned producer learns to stop.
The client reads with message().await, which returns Result<Option<T>, Status>. None means the server finished cleanly.

Example 3: Client streaming

The server receives a Streaming<Order> and reads it in a loop. It answers once after the client half-closes.

src/import.rs

Rust

use tonic::{Request, Response, Status, Streaming};

use crate::pb::{ImportSummary, Order};

pub async fn import_orders(
    request: Request<Streaming<Order>>,
) -> Result<Response<ImportSummary>, Status> {
    let mut stream = request.into_inner();
    let mut imported = 0;

    while let Some(order) = stream.message().await? {
        if order.id <= 0 {
            return Err(Status::invalid_argument("order id must be positive"));
        }
        imported += 1;
    }

    Ok(Response::new(ImportSummary { imported }))
}

The server processes each message as it arrives instead of buffering the batch, and it can reject the call early. The client passes any stream of messages: an iterator for a fixed batch, a channel receiver for a live feed.

Example 4: Bidirectional streaming

The two directions are independent. This handler answers each inbound message, but a server could answer in batches or send without receiving. Order is preserved within one direction and is undefined between directions.

src/chat.rs

Rust

use tokio_stream::wrappers::ReceiverStream;
use tonic::{Request, Response, Status, Streaming};

use crate::pb::ChatMessage;
use crate::BoxStream;

pub async fn chat(
    request: Request<Streaming<ChatMessage>>,
) -> Result<Response<BoxStream<ChatMessage>>, Status> {
    let mut inbound = request.into_inner();
    let (tx, rx) = tokio::sync::mpsc::channel(8);

    tokio::spawn(async move {
        while let Ok(Some(message)) = inbound.message().await {
            let reply = ChatMessage {
                text: format!("echo: {}", message.text),
            };
            if tx.send(Ok(reply)).await.is_err() {
                break;
            }
        }
    });

    Ok(Response::new(
        Box::pin(ReceiverStream::new(rx)) as BoxStream<ChatMessage>
    ))
}

A stream keeps a connection open, pins its messages to one backend, and cannot be retried as a whole once messages were sent. Unary calls fit single request-response exchanges. Streams fit data that is a sequence.

Example 5: Status codes and metadata

A Status has a code from a fixed set and a message. Callers react to the code without parsing text.

Code	Use for	Retry
`InvalidArgument`	The request is malformed regardless of state	No
`NotFound`	The entity does not exist	No
`AlreadyExists`	A create conflicts with an existing entity	No
`PermissionDenied`	The caller is known but not allowed	No
`Unauthenticated`	Credentials are missing or invalid	After fixing credentials
`FailedPrecondition`	The system state does not allow the call	No, until state changes
`ResourceExhausted`	Quota or rate limit reached	After backoff
`Aborted`	Concurrency conflict, such as a failed optimistic check	Yes, at a higher level
`Unavailable`	The service cannot serve now	Yes, with backoff
`DeadlineExceeded`	The deadline passed	Only if the call is idempotent
`Unimplemented`	The method is not supported	No
`Internal`	A bug or broken invariant	No

Tonic has one constructor per code, such as Status::not_found(...). NotFound corresponds to HTTP 404, InvalidArgument to 400, Unauthenticated to 401, PermissionDenied to 403, and Unavailable to 503. The resilience patterns retry rules depend on these codes. Retrying a non-idempotent call after DeadlineExceeded can run it twice, because the server may finish after the client gave up. See SQL transactions for idempotency keys.

Metadata carries information about the call, such as tokens and request ids. Keys are lowercase ASCII. Values are ASCII strings, except keys ending in -bin, which hold bytes.

The handler below reads x-request-id from metadata and a CallerId that the interceptor in Example 6 stores in the request extensions. Order 2 sleeps for 2 seconds so that Example 7 can show a deadline.

src/unary.rs

Rust

use std::time::Duration;

use tonic::{Request, Response, Status};

use crate::auth::CallerId;
use crate::pb::{GetOrderRequest, Order};

pub async fn get_order(request: Request<GetOrderRequest>) -> Result<Response<Order>, Status> {
    let request_id = request
        .metadata()
        .get("x-request-id")
        .and_then(|value| value.to_str().ok())
        .unwrap_or("none")
        .to_owned();
    let caller = request.extensions().get::<CallerId>().map(|caller| caller.0);
    let order_id = request.into_inner().order_id;

    match order_id {
        id if id <= 0 => Err(Status::invalid_argument("order id must be positive")),
        1 => Ok(Response::new(Order {
            id: 1,
            status: format!("paid; request_id={request_id}; caller={caller:?}"),
            total_cents: 1250,
        })),
        2 => {
            tokio::time::sleep(Duration::from_secs(2)).await;
            Ok(Response::new(Order {
                id: 2,
                status: "shipped".to_owned(),
                total_cents: 830,
            }))
        }
        id => Err(Status::not_found(format!("order {id} not found"))),
    }
}

Propagate the request id and trace context from metadata into the current span, as in observability.

Example 6: An authentication interceptor and the server

An interceptor is a function FnMut(Request<()>) -> Result<Request<()>, Status>. It sees metadata and extensions but not the typed message. It returns the request to continue or a Status to reject the call.

src/auth.rs

Rust

use tonic::{Request, Status};

#[derive(Clone, Copy, Debug)]
pub struct CallerId(pub u64);

pub fn check_auth(mut request: Request<()>) -> Result<Request<()>, Status> {
    let token = request
        .metadata()
        .get("authorization")
        .and_then(|value| value.to_str().ok());

    match token {
        Some("Bearer secret-token") => {
            request.extensions_mut().insert(CallerId(42));
            Ok(request)
        }
        _ => Err(Status::unauthenticated("missing or invalid token")),
    }
}

The fixed token only shows the mechanism. Real validation uses the JWT checks from OAuth, JWT, and mTLS.

src/main.rs

Rust

use orders::{auth::check_auth, pb::orders_server::OrdersServer, service::OrdersService};
use tonic::transport::Server;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "127.0.0.1:50052".parse()?;
    println!("orders listening on {addr}");
    Server::builder()
        .add_service(OrdersServer::with_interceptor(OrdersService, check_auth))
        .serve(addr)
        .await?;
    Ok(())
}

Interceptors cannot read messages or responses. Timing, result logging, and response changes use a Tower layer, which wraps the whole service. Tonic servers are Tower services, so the layers from Tower middleware attach with Server::builder().layer(...).

Example 7: Deadlines, cancellation, and the client

A deadline is the point in time by which a call must finish. A timeout is the duration that produces it. Request::set_timeout sends the duration in the grpc-timeout header so the server knows how long the caller waits.

Propagation. If a gateway allows 500 ms and Orders calls Catalog with a fresh 5 s timeout, Catalog keeps working after the gateway gave up. Each hop passes the remaining budget, computed from the time already spent, to the next call.

Cancellation happens when the caller drops the call future, the deadline passes, or the connection closes. The server drops the handler future at its next .await.

Code between .await points always runs to completion.
A task started with tokio::spawn is not cancelled with the handler. The producers in Examples 2 and 4 stop only because tx.send fails.
A database transaction held across an .await is rolled back when the handler is dropped. Work that must survive cancellation belongs in a spawned task or in the outbox pattern.

The client below guards the slow call with a local tokio::time::timeout of 300 ms, which drops the call future and cancels it. The function add_token is a client interceptor that attaches the token to every call.

src/bin/client.rs

Rust

use std::time::Duration;

use orders::pb::{
    orders_client::OrdersClient, ChatMessage, GetOrderRequest, ImportSummary, Order,
    WatchOrderRequest,
};
use tonic::{transport::Channel, Request, Status};

fn add_token(mut request: Request<()>) -> Result<Request<()>, Status> {
    request
        .metadata_mut()
        .insert("authorization", "Bearer secret-token".parse().unwrap());
    Ok(request)
}

fn sample_order(id: i64) -> Order {
    Order {
        id,
        status: "new".to_owned(),
        total_cents: 100,
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let channel = Channel::from_static("http://127.0.0.1:50052").connect().await?;

    let mut authed = OrdersClient::with_interceptor(channel.clone(), add_token);
    let mut anonymous = OrdersClient::new(channel);

    // Unary with metadata.
    let mut request = Request::new(GetOrderRequest { order_id: 1 });
    request
        .metadata_mut()
        .insert("x-request-id", "req-1042".parse().unwrap());
    let first = authed.get_order(request).await?.into_inner();
    println!("order {}: {}", first.id, first.status);

    // Interceptor rejects a call without a token.
    if let Err(status) = anonymous.get_order(GetOrderRequest { order_id: 1 }).await {
        println!("{:?}: {}", status.code(), status.message());
    }

    // Status code from the handler.
    if let Err(status) = authed.get_order(GetOrderRequest { order_id: 99 }).await {
        println!("{:?}: {}", status.code(), status.message());
    }

    // Deadline: the server sleeps 2 s, the caller stops waiting after 300 ms.
    let mut slow = Request::new(GetOrderRequest { order_id: 2 });
    slow.set_timeout(Duration::from_secs(1));
    let guarded = tokio::time::timeout(Duration::from_millis(300), authed.get_order(slow)).await;
    if guarded.is_err() {
        println!("slow order dropped after 300 ms");
    }

    // Server streaming.
    let mut events = authed
        .watch_order(WatchOrderRequest { order_id: 7 })
        .await?
        .into_inner();
    while let Some(event) = events.message().await? {
        println!("order {} is {}", event.order_id, event.status);
    }

    // Client streaming.
    let summary: ImportSummary = authed
        .import_orders(tokio_stream::iter(vec![sample_order(1), sample_order(2)]))
        .await?
        .into_inner();
    println!("imported {}", summary.imported);

    if let Err(status) = authed
        .import_orders(tokio_stream::iter(vec![sample_order(3), sample_order(0)]))
        .await
    {
        println!("{:?}: {}", status.code(), status.message());
    }

    // Bidirectional streaming.
    let outbound = tokio_stream::iter(["hello", "status", "bye"].map(|text| ChatMessage {
        text: text.to_owned(),
    }));
    let mut replies = authed.chat(outbound).await?.into_inner();
    while let Some(reply) = replies.message().await? {
        println!("{}", reply.text);
    }
    Ok(())
}

Complete example

Prerequisites: Rust, protoc. File tree:

Text

orders/
├── Cargo.toml
├── build.rs
├── proto/
│   └── orders.proto
└── src/
    ├── lib.rs
    ├── service.rs
    ├── watch.rs
    ├── import.rs
    ├── chat.rs
    ├── unary.rs
    ├── auth.rs
    ├── main.rs
    └── bin/
        └── client.rs

Cargo.toml

TOML

[package]
name = "orders"
version = "0.1.0"
edition = "2021"

[dependencies]
tonic = "0.13"
prost = "0.13"
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "sync"] }
tokio-stream = "0.1"

[build-dependencies]
tonic-build = "0.13"

build.rs

Rust

fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_build::compile_protos("proto/orders.proto")?;
    Ok(())
}

proto/orders.proto

PROTOBUF

syntax = "proto3";

package storefront.orders.v1;

service Orders {
  // Unary
  rpc GetOrder(GetOrderRequest) returns (Order);
  // Server streaming: one request, many responses
  rpc WatchOrder(WatchOrderRequest) returns (stream OrderEvent);
  // Client streaming: many requests, one response
  rpc ImportOrders(stream Order) returns (ImportSummary);
  // Bidirectional streaming
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message GetOrderRequest {
  int64 order_id = 1;
}

message Order {
  int64 id = 1;
  string status = 2;
  int64 total_cents = 3;
}

message WatchOrderRequest {
  int64 order_id = 1;
}

message OrderEvent {
  int64 order_id = 1;
  string status = 2;
}

message ImportSummary {
  int32 imported = 1;
}

message ChatMessage {
  string text = 1;
}

src/lib.rs

Rust

use std::pin::Pin;

use tokio_stream::Stream;
use tonic::Status;

pub mod pb {
    tonic::include_proto!("storefront.orders.v1");
}

pub mod auth;
pub mod chat;
pub mod import;
pub mod service;
pub mod unary;
pub mod watch;

pub type BoxStream<T> = Pin<Box<dyn Stream<Item = Result<T, Status>> + Send>>;

src/service.rs

Rust

use tonic::{Request, Response, Status, Streaming};

use crate::pb::{
    orders_server::Orders, ChatMessage, GetOrderRequest, ImportSummary, Order, OrderEvent,
    WatchOrderRequest,
};
use crate::{chat, import, unary, watch, BoxStream};

#[derive(Default)]
pub struct OrdersService;

#[tonic::async_trait]
impl Orders for OrdersService {
    type WatchOrderStream = BoxStream<OrderEvent>;
    type ChatStream = BoxStream<ChatMessage>;

    async fn get_order(
        &self,
        request: Request<GetOrderRequest>,
    ) -> Result<Response<Order>, Status> {
        unary::get_order(request).await
    }

    async fn watch_order(
        &self,
        request: Request<WatchOrderRequest>,
    ) -> Result<Response<Self::WatchOrderStream>, Status> {
        watch::watch_order(request).await
    }

    async fn import_orders(
        &self,
        request: Request<Streaming<Order>>,
    ) -> Result<Response<ImportSummary>, Status> {
        import::import_orders(request).await
    }

    async fn chat(
        &self,
        request: Request<Streaming<ChatMessage>>,
    ) -> Result<Response<Self::ChatStream>, Status> {
        chat::chat(request).await
    }
}

src/watch.rs

Rust

use tokio_stream::wrappers::ReceiverStream;
use tonic::{Request, Response, Status};

use crate::pb::{OrderEvent, WatchOrderRequest};
use crate::BoxStream;

pub async fn watch_order(
    request: Request<WatchOrderRequest>,
) -> Result<Response<BoxStream<OrderEvent>>, Status> {
    let order_id = request.into_inner().order_id;
    if order_id <= 0 {
        return Err(Status::invalid_argument("order id must be positive"));
    }

    let (tx, rx) = tokio::sync::mpsc::channel(4);
    tokio::spawn(async move {
        for status in ["placed", "paid", "shipped"] {
            let event = OrderEvent {
                order_id,
                status: status.to_owned(),
            };
            if tx.send(Ok(event)).await.is_err() {
                // The client disconnected or cancelled: stop producing.
                break;
            }
        }
    });

    Ok(Response::new(
        Box::pin(ReceiverStream::new(rx)) as BoxStream<OrderEvent>
    ))
}

src/import.rs

Rust

use tonic::{Request, Response, Status, Streaming};

use crate::pb::{ImportSummary, Order};

pub async fn import_orders(
    request: Request<Streaming<Order>>,
) -> Result<Response<ImportSummary>, Status> {
    let mut stream = request.into_inner();
    let mut imported = 0;

    while let Some(order) = stream.message().await? {
        if order.id <= 0 {
            return Err(Status::invalid_argument("order id must be positive"));
        }
        imported += 1;
    }

    Ok(Response::new(ImportSummary { imported }))
}

src/chat.rs

Rust

use tokio_stream::wrappers::ReceiverStream;
use tonic::{Request, Response, Status, Streaming};

use crate::pb::ChatMessage;
use crate::BoxStream;

pub async fn chat(
    request: Request<Streaming<ChatMessage>>,
) -> Result<Response<BoxStream<ChatMessage>>, Status> {
    let mut inbound = request.into_inner();
    let (tx, rx) = tokio::sync::mpsc::channel(8);

    tokio::spawn(async move {
        while let Ok(Some(message)) = inbound.message().await {
            let reply = ChatMessage {
                text: format!("echo: {}", message.text),
            };
            if tx.send(Ok(reply)).await.is_err() {
                break;
            }
        }
    });

    Ok(Response::new(
        Box::pin(ReceiverStream::new(rx)) as BoxStream<ChatMessage>
    ))
}

src/unary.rs

Rust

use std::time::Duration;

use tonic::{Request, Response, Status};

use crate::auth::CallerId;
use crate::pb::{GetOrderRequest, Order};

pub async fn get_order(request: Request<GetOrderRequest>) -> Result<Response<Order>, Status> {
    let request_id = request
        .metadata()
        .get("x-request-id")
        .and_then(|value| value.to_str().ok())
        .unwrap_or("none")
        .to_owned();
    let caller = request.extensions().get::<CallerId>().map(|caller| caller.0);
    let order_id = request.into_inner().order_id;

    match order_id {
        id if id <= 0 => Err(Status::invalid_argument("order id must be positive")),
        1 => Ok(Response::new(Order {
            id: 1,
            status: format!("paid; request_id={request_id}; caller={caller:?}"),
            total_cents: 1250,
        })),
        2 => {
            tokio::time::sleep(Duration::from_secs(2)).await;
            Ok(Response::new(Order {
                id: 2,
                status: "shipped".to_owned(),
                total_cents: 830,
            }))
        }
        id => Err(Status::not_found(format!("order {id} not found"))),
    }
}

src/auth.rs

Rust

use tonic::{Request, Status};

#[derive(Clone, Copy, Debug)]
pub struct CallerId(pub u64);

pub fn check_auth(mut request: Request<()>) -> Result<Request<()>, Status> {
    let token = request
        .metadata()
        .get("authorization")
        .and_then(|value| value.to_str().ok());

    match token {
        Some("Bearer secret-token") => {
            request.extensions_mut().insert(CallerId(42));
            Ok(request)
        }
        _ => Err(Status::unauthenticated("missing or invalid token")),
    }
}

src/main.rs

Rust

use orders::{auth::check_auth, pb::orders_server::OrdersServer, service::OrdersService};
use tonic::transport::Server;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "127.0.0.1:50052".parse()?;
    println!("orders listening on {addr}");
    Server::builder()
        .add_service(OrdersServer::with_interceptor(OrdersService, check_auth))
        .serve(addr)
        .await?;
    Ok(())
}

src/bin/client.rs

Rust

use std::time::Duration;

use orders::pb::{
    orders_client::OrdersClient, ChatMessage, GetOrderRequest, ImportSummary, Order,
    WatchOrderRequest,
};
use tonic::{transport::Channel, Request, Status};

fn add_token(mut request: Request<()>) -> Result<Request<()>, Status> {
    request
        .metadata_mut()
        .insert("authorization", "Bearer secret-token".parse().unwrap());
    Ok(request)
}

fn sample_order(id: i64) -> Order {
    Order {
        id,
        status: "new".to_owned(),
        total_cents: 100,
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let channel = Channel::from_static("http://127.0.0.1:50052").connect().await?;

    let mut authed = OrdersClient::with_interceptor(channel.clone(), add_token);
    let mut anonymous = OrdersClient::new(channel);

    // Unary with metadata.
    let mut request = Request::new(GetOrderRequest { order_id: 1 });
    request
        .metadata_mut()
        .insert("x-request-id", "req-1042".parse().unwrap());
    let first = authed.get_order(request).await?.into_inner();
    println!("order {}: {}", first.id, first.status);

    // Interceptor rejects a call without a token.
    if let Err(status) = anonymous.get_order(GetOrderRequest { order_id: 1 }).await {
        println!("{:?}: {}", status.code(), status.message());
    }

    // Status code from the handler.
    if let Err(status) = authed.get_order(GetOrderRequest { order_id: 99 }).await {
        println!("{:?}: {}", status.code(), status.message());
    }

    // Deadline: the server sleeps 2 s, the caller stops waiting after 300 ms.
    let mut slow = Request::new(GetOrderRequest { order_id: 2 });
    slow.set_timeout(Duration::from_secs(1));
    let guarded = tokio::time::timeout(Duration::from_millis(300), authed.get_order(slow)).await;
    if guarded.is_err() {
        println!("slow order dropped after 300 ms");
    }

    // Server streaming.
    let mut events = authed
        .watch_order(WatchOrderRequest { order_id: 7 })
        .await?
        .into_inner();
    while let Some(event) = events.message().await? {
        println!("order {} is {}", event.order_id, event.status);
    }

    // Client streaming.
    let summary: ImportSummary = authed
        .import_orders(tokio_stream::iter(vec![sample_order(1), sample_order(2)]))
        .await?
        .into_inner();
    println!("imported {}", summary.imported);

    if let Err(status) = authed
        .import_orders(tokio_stream::iter(vec![sample_order(3), sample_order(0)]))
        .await
    {
        println!("{:?}: {}", status.code(), status.message());
    }

    // Bidirectional streaming.
    let outbound = tokio_stream::iter(["hello", "status", "bye"].map(|text| ChatMessage {
        text: text.to_owned(),
    }));
    let mut replies = authed.chat(outbound).await?.into_inner();
    while let Some(reply) = replies.message().await? {
        println!("{}", reply.text);
    }
    Ok(())
}

Run the server in one terminal and the client in another:

Terminal

cargo run --bin orders
cargo run --bin client

Server output:

Output

orders listening on 127.0.0.1:50052

Client output:

Output

order 1: paid; request_id=req-1042; caller=Some(42)
Unauthenticated: missing or invalid token
NotFound: order 99 not found
slow order dropped after 300 ms
order 7 is placed
order 7 is paid
order 7 is shipped
imported 2
InvalidArgument: order id must be positive
echo: hello
echo: status
echo: bye

curl does not speak gRPC framing, so the Rust client is the test tool here. Reflection and grpcurl come in gRPC in production.

Summary

A .proto rpc uses stream on the request, the response, or both to define streaming calls.
Server streaming returns a boxed stream, usually built from an mpsc receiver. Client streaming and bidirectional handlers read a Streaming<T> with message().await.
Bounded channels plus HTTP/2 flow control give backpressure. A failed send tells a producer to stop.
Status codes classify failures, and each code implies a retry policy. Unavailable is the standard retryable code.
Metadata carries tokens and trace ids. Keys are lowercase, and binary values use the -bin suffix.
grpc-timeout carries the deadline. Each hop passes the remaining time, not a fresh timeout.
Dropping a handler future cancels it at the next .await. Spawned tasks need their own stop condition.
Interceptors see metadata and extensions only. Tower layers see whole requests and responses.