### Start tarpc Server and Client

Source: https://github.com/google/tarpc/blob/main/tarpc/README.md

This main function sets up an in-process channel, starts the server, and then connects a client to it. The client then calls the 'hello' RPC.

```rust
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let (client_transport, server_transport) = tarpc::transport::channel::unbounded();

    let server = server::BaseChannel::with_defaults(server_transport);
    tokio::spawn(
        server.execute(HelloServer.serve())
            // Handle all requests concurrently.
            .for_each(|response| async move {
                tokio::spawn(response);
            }));

    // WorldClient is generated by the #[tarpc::service] attribute. It has a constructor `new`
    // that takes a config and any Transport as input.
    let mut client = WorldClient::new(client::Config::default(), client_transport).spawn();

    // The client has an RPC method for each RPC defined in the annotated trait. It takes the same
    // args as defined, with the addition of a Context, which is always the first arg. The Context
    // specifies a deadline and trace information which can be helpful in debugging requests.
    let hello = client.hello(context::current(), "Stim".to_string()).await?;

    println!("{hello}");

    Ok(())
}
```

--------------------------------

### Run Tarpc Server

Source: https://github.com/google/tarpc/blob/main/example-service/README.md

Starts the tarpc server. Ensure RUST_LOG is set to trace to view Jaeger traces.

```bash
cargo run --bin server -- --port 50051
```

--------------------------------

### Define tarpc Service

Source: https://github.com/google/tarpc/blob/main/tarpc/README.md

Define your RPC service using the #[tarpc::service] macro. This example defines a simple 'hello' RPC.

```rust
use futures::prelude::*;
use tarpc::{
    client,
    context,
    server::{self, Channel},
};

// This is the service definition. It looks a lot like a trait definition.
// It defines one RPC, hello, which takes one arg, name, and returns a String.
#[tarpc::service]
trait World {
    /// Returns a greeting for name.
    async fn hello(name: String) -> String;
}
```

--------------------------------

### Defining an RPC Service with `#[tarpc::service]`

Source: https://context7.com/google/tarpc/llms.txt

This example demonstrates how to define an RPC service using the `#[tarpc::service]` attribute macro on a Rust trait. It generates a client stub and a server trait, enabling the creation of both client and server components for the RPC service.

```APIDOC
## `#[tarpc::service]` — Define an RPC service

Annotating a trait with `#[tarpc::service]` generates: a server trait to implement, a `serve()` method that turns an impl into a `Serve` handler, and a `<TraitName>Client` stub with one async method per RPC.

```rust
use futures::prelude::*;
use tarpc::{client, context, server::{self, Channel}};

// Service definition — generates WorldClient and the World trait
#[tarpc::service]
trait World {
    /// Returns a greeting for the given name.
    async fn hello(name: String) -> String;
    /// Adds two integers.
    async fn add(a: i64, b: i64) -> i64;
}

// Server implementation
#[derive(Clone)]
struct MyServer;

impl World for MyServer {
    async fn hello(self, _ctx: context::Context, name: String) -> String {
        format!("Hello, {name}!")
    }
    async fn add(self, _ctx: context::Context, a: i64, b: i64) -> i64 {
        a + b
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // In-process channel transport (no serialization cost)
    let (client_transport, server_transport) = tarpc::transport::channel::unbounded();

    let server = server::BaseChannel::with_defaults(server_transport);
    tokio::spawn(
        server.execute(MyServer.serve())
            .for_each(|response| async move { tokio::spawn(response); }),
    );

    let mut client = WorldClient::new(client::Config::default(), client_transport).spawn();

    let greeting = client.hello(context::current(), "Alice".to_string()).await?;
    println!("{greeting}"); // Hello, Alice!

    let sum = client.add(context::current(), 3, 4).await?;
    println!("{sum}"); // 7

    Ok(())
}
```
```

--------------------------------

### Run Tarpc Client

Source: https://github.com/google/tarpc/blob/main/example-service/README.md

Starts the tarpc client, connecting to the specified server address and setting a name. Ensure RUST_LOG is set to trace to view Jaeger traces.

```bash
cargo run --bin client -- --server-addr "[::1]:50051" --name "Bob"
```

--------------------------------

### Define RPC Service with `#[tarpc::service]`

Source: https://context7.com/google/tarpc/llms.txt

Use `#[tarpc::service]` on a Rust trait to automatically generate server implementations and client stubs. This example demonstrates defining a 'World' service with 'hello' and 'add' RPCs and setting up an in-process client/server.

```rust
use futures::prelude::*;
use tarpc::{client, context, server::{self, Channel}};

// Service definition — generates WorldClient and the World trait
#[tarpc::service]
trait World {
    /// Returns a greeting for the given name.
    async fn hello(name: String) -> String;
    /// Adds two integers.
    async fn add(a: i64, b: i64) -> i64;
}

// Server implementation
#[derive(Clone)]
struct MyServer;

impl World for MyServer {
    async fn hello(self, _ctx: context::Context, name: String) -> String {
        format!("Hello, {name}!")
    }
    async fn add(self, _ctx: context::Context, a: i64, b: i64) -> i64 {
        a + b
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // In-process channel transport (no serialization cost)
    let (client_transport, server_transport) = tarpc::transport::channel::unbounded();

    let server = server::BaseChannel::with_defaults(server_transport);
    tokio::spawn(
        server.execute(MyServer.serve())
            .for_each(|response| async move { tokio::spawn(response); })
    );

    let mut client = WorldClient::new(client::Config::default(), client_transport).spawn();

    let greeting = client.hello(context::current(), "Alice".to_string()).await?;
    println!("{greeting}"); // Hello, Alice!

    let sum = client.add(context::current(), 3, 4).await?;
    println!("{sum}"); // 7

    Ok(())
}
```

--------------------------------

### Initialize OpenTelemetry Tracing

Source: https://context7.com/google/tarpc/llms.txt

Sets up OpenTelemetry tracing with a specified service name and an OTLP span exporter. This function should be called once during application initialization. It configures the global tracer provider and initializes the tracing subscriber with OpenTelemetry and other layers.

```rust
use opentelemetry::trace::TracerProvider as _;
use tracing_subscriber::{fmt::format::FmtSpan, prelude::*};

fn init_tracing(service_name: &'static str)
    -> anyhow::Result<opentelemetry_sdk::trace::SdkTracerProvider>
{
    let tracer_provider = opentelemetry_sdk::trace::SdkTracerProvider::builder()
        .with_resource(
            opentelemetry_sdk::Resource::builder()
                .with_service_name(service_name)
                .build(),
        )
        .with_batch_exporter(
            opentelemetry_otlp::SpanExporter::builder()
                .with_tonic()
                .build()?,
        )
        .build();
    opentelemetry::global::set_tracer_provider(tracer_provider.clone());
    let tracer = tracer_provider.tracer(service_name);

    tracing_subscriber::registry()
        .with(tracing_subscriber::EnvFilter::from_default_env())
        .with(tracing_subscriber::fmt::layer().with_span_events(FmtSpan::NEW | FmtSpan::CLOSE))
        .with(tracing_opentelemetry::layer().with_tracer(tracer))
        .try_init()?;

    Ok(tracer_provider)
}

// After calling init_tracing, all tarpc calls will emit spans visible in Jaeger/Tempo/etc.
// The trace_id in context::current() propagates through the entire call chain.
// Expected output in tracing backend:
//   Span: RPC hello [client]
//     → Span: RPC hello [server]
//         otel.name = "hello"
//         rpc.trace_id = "a1b2c3..."
//         rpc.deadline = "2025-01-01T00:00:10Z"
```

--------------------------------

### Connect and Listen with Unix Domain Sockets

Source: https://context7.com/google/tarpc/llms.txt

Use `serde_transport::unix::connect` and `listen` for low-latency local IPC via Unix Domain Sockets. Requires `serde-transport` and `unix` features. The API is identical to TCP.

```rust
use futures::prelude::*;
use tarpc::{
    client,
    context,
    server::{self, BaseChannel, Channel, serve},
    serde_transport::unix,
};
use tokio_serde::formats::SymmetricalJson;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // TempPathBuf creates a socket file and removes it on drop
    let sock = unix::TempPathBuf::with_random("my_service");

    let mut listener = unix::listen(&sock, SymmetricalJson::<i32>::default).await?;
    tokio::spawn(async move {
        if let Some(Ok(transport)) = listener.next().await {
            BaseChannel::with_defaults(transport)
                .execute(serve(|_ctx, n: i32| async move { Ok(n + 1) }))
                .for_each(|f| async move { tokio::spawn(f); })
                .await;
        }
    });

    let transport = unix::connect(&sock, SymmetricalJson::<i32>::default).await?;
    let client = client::new(client::Config::default(), transport).spawn();
    let result = client.call(context::current(), 41).await?;
    println!("{result}"); // 42

    Ok(())
}
```

--------------------------------

### Using `context::current()` and `context::Context`

Source: https://context7.com/google/tarpc/llms.txt

This section explains how to use `context::Context` for RPC calls, which includes deadline propagation and tracing information. Servers can inspect the context for custom logic and automatically cease work when a deadline passes.

```APIDOC
## `context::current()` / `context::Context` — Request context with deadline and trace

Every RPC call takes a `context::Context` as its first argument. The context carries a deadline (defaulting to 10 seconds from now) and an OpenTelemetry-compatible trace context. Servers automatically cease work when the deadline passes. Downstream requests that use the propagated context inherit the remaining deadline.

```rust
use tarpc::context;
use std::time::{Duration, Instant};

// Default context: deadline 10s from now, inherits current tracing span
let ctx = context::current();

// Custom deadline: 500ms from now
let mut short_ctx = context::current();
short_ctx.deadline = Instant::now() + Duration::from_millis(500);

// Inspect the trace ID (useful for logging correlated distributed traces)
println!("trace_id = {}", ctx.trace_id());

// On the server side, the context is the first argument after `self`:
#[tarpc::service]
trait Compute {
    async fn fib(n: u32) -> u64;
}

#[derive(Clone)]
struct ComputeServer;

impl Compute for ComputeServer {
    async fn fib(self, ctx: context::Context, n: u32) -> u64 {
        // ctx.deadline is already enforced by the framework;
        // you can also inspect it for your own logic
        println!("deadline remaining: {:?}", ctx.deadline.checked_duration_since(std::time::Instant::now()));
        (0..n).fold((0u64, 1u64), |(a, b), _| (b, a + b)).0
    }
}
```
```

--------------------------------

### server::serve

Source: https://context7.com/google/tarpc/llms.txt

Wrap a closure as a `Serve` handler. `server::serve(f)` turns any `FnOnce(Context, Req) -> impl Future<Output = Result<Resp, ServerError>>` into a `Serve` implementor, avoiding the need to write a struct + impl.

```APIDOC
## `server::serve` — Wrap a closure as a `Serve` handler

`server::serve(f)` turns any `FnOnce(Context, Req) -> impl Future<Output = Result<Resp, ServerError>>` into a `Serve` implementor, avoiding the need to write a struct + impl.

### Usage

- `serve(handler_closure)`: Creates a `Serve` implementor from a closure that takes a `context::Context` and a request, returning a future that resolves to a `Result`.
- The returned handler can be called directly using `.serve(context, request)`.
```

--------------------------------

### client::Config

Source: https://context7.com/google/tarpc/llms.txt

Configure client-side concurrency and buffering. This struct is accepted by the generated `<ServiceName>Client::new(config, transport)`.

```APIDOC
## `client::Config` — Configure client-side concurrency and buffering

`client::Config` controls how many requests can be in-flight simultaneously and the size of the internal send buffer. The generated `<ServiceName>Client::new(config, transport)` accepts this struct.

### Fields

- **max_in_flight_requests** (usize) - Allows specifying the maximum number of concurrent in-flight requests.
- **pending_request_buffer** (usize) - Sets the size of the internal send buffer before applying back-pressure.
```

--------------------------------

### Configure tarpc client concurrency and buffering

Source: https://context7.com/google/tarpc/llms.txt

Use `client::Config` to set the maximum number of in-flight requests and the size of the internal send buffer. This configuration is passed when creating a new client.

```rust
use tarpc::{client, context, server::{self, Channel}, transport};

#[tarpc::service] 
trait Echo {
    async fn echo(msg: String) -> String;
}

#[derive(Clone)] 
struct EchoServer;
impl Echo for EchoServer {
    async fn echo(self, _: context::Context, msg: String) -> String {
        msg
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let (client_tx, server_rx) = transport::channel::unbounded();

    tokio::spawn(
        server::BaseChannel::with_defaults(server_rx)
            .execute(EchoServer.serve())
            .for_each(|f| async move { tokio::spawn(f); }),
    );

    let config = client::Config {
        // Allow up to 500 concurrent in-flight requests
        max_in_flight_requests: 500,
        // Buffer up to 50 requests before applying back-pressure
        pending_request_buffer: 50,
    };

    let client = EchoClient::new(config, client_tx).spawn();

    // Fire off 10 concurrent requests
    let futs: Vec<_> = (0..10)
        .map(|i| client.echo(context::current(), format!("msg-{i}")))
        .collect();
    let results = futures::future::join_all(futs).await;
    for r in results {
        println!("{}", r?);
    }

    Ok(())
}
```

--------------------------------

### Wrap a closure as a Serve handler with server::serve

Source: https://context7.com/google/tarpc/llms.txt

Use `server::serve(f)` to turn any `FnOnce(Context, Req) -> impl Future<Output = Result<Resp, ServerError>>` into a `Serve` implementor. This avoids the need to write a struct and impl block.

```rust
use tarpc::{
    ServerError,
    context,
    server::{serve, Serve},
};
use std::io;

// A simple handler that rejects negative numbers
let handler = serve(|_ctx: context::Context, n: i32| async move {
    if n < 0 {
        Err(ServerError::new(io::ErrorKind::InvalidInput, format!("{n} is negative")))
    } else {
        Ok(n * n)
    }
});

// Call it directly in tests
let result = handler.serve(context::current(), 5).await;
assert_eq!(result.unwrap(), 25);

let err = serve(|_ctx: context::Context, n: i32| async move {
    if n < 0 {
        Err(ServerError::new(io::ErrorKind::InvalidInput, "negative".into()))
    } else {
        Ok(n)
    }
}).serve(context::current(), -1).await;
assert!(err.is_err());
```

--------------------------------

### Add tarpc and tokio Dependencies

Source: https://github.com/google/tarpc/blob/main/tarpc/README.md

Add these dependencies to your Cargo.toml file to use tarpc with tokio.

```toml
anyhow = "1.0"
futures = "0.3"
tarpc = { version = "0.37", features = ["tokio1"] }
tokio = { version = "1.0", features = ["rt-multi-thread", "macros"] }
```

--------------------------------

### High-level concurrent request dispatch with Channel::execute

Source: https://context7.com/google/tarpc/llms.txt

Use `Channel::execute(serve_fn)` for the simplest way to serve all requests concurrently. It returns a `Stream` of response futures that must be spawned or awaited.

```rust
use tarpc::{client, context, server::{self, BaseChannel, Channel, serve}, transport};
use futures::prelude::*;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let (tx, rx) = transport::channel::unbounded();
    let client = client::new(client::Config::default(), tx).spawn();

    tokio::spawn(
        BaseChannel::with_defaults(rx)
            .execute(serve(|_ctx, s: String| async move {
                Ok(s.to_uppercase())
            }))
            // Spawn each response future concurrently
            .for_each(|response_fut| async move {
                tokio::spawn(response_fut);
            }),
    );

    let resp = client.call(context::current(), "hello".to_string()).await?;
    println!("{resp}"); // HELLO

    Ok(())
}
```

--------------------------------

### RPC Request Context with Deadline and Tracing

Source: https://context7.com/google/tarpc/llms.txt

RPC calls use `context::Context` for deadlines and trace propagation. Servers enforce deadlines, and downstream requests inherit remaining deadlines. Inspect trace IDs for correlated logging.

```rust
use tarpc::context;
use std::time::{Duration, Instant};

// Default context: deadline 10s from now, inherits current tracing span
let ctx = context::current();

// Custom deadline: 500ms from now
let mut short_ctx = context::current();
short_ctx.deadline = Instant::now() + Duration::from_millis(500);

// Inspect the trace ID (useful for logging correlated distributed traces)
println!("trace_id = {}", ctx.trace_id());

// On the server side, the context is the first argument after `self`:
#[tarpc::service]
trait Compute {
    async fn fib(n: u32) -> u64;
}

#[derive(Clone)]
struct ComputeServer;

impl Compute for ComputeServer {
    async fn fib(self, ctx: context::Context, n: u32) -> u64 {
        // ctx.deadline is already enforced by the framework;
        // you can also inspect it for your own logic
        println!("deadline remaining: {:?}", ctx.deadline.checked_duration_since(std::time::Instant::now()));
        (0..n).fold((0u64, 1u64), |(a, b), _| (b, a + b)).0
    }
}
```

--------------------------------

### Execute and Spawn Request Handlers in tarpc

Source: https://github.com/google/tarpc/blob/main/RELEASES.md

This snippet demonstrates how to execute request handlers and spawn them in separate tasks using `tokio::spawn`. This pattern is used to replicate the behavior of the deprecated `Channel::execute` which previously handled spawning.

```rust
channel.execute(server.serve())
       .for_each(|rpc| { tokio::spawn(rpc); })

```

--------------------------------

### serde_transport::tcp::connect / listen

Source: https://context7.com/google/tarpc/llms.txt

Provides network transport over TCP using Serde for serialization. Includes `connect` and `listen` helpers for TCP streams with length-delimited Serde codec.

```APIDOC
## `serde_transport::tcp::connect` / `listen` — Network transport over TCP with Serde

Requires features `serde-transport` and `tcp`. Provides `connect` and `listen` helpers that wrap TCP streams in a length-delimited Serde codec transport, ready to plug into tarpc's client/server APIs.

### Example: Setting up a TCP server and client with Serde JSON
```rust
use futures::prelude::*;
use serde::{Deserialize, Serialize};
use tarpc::{
    client,
    context,
    server::{self, BaseChannel, Channel},
    serde_transport::tcp,
};
use tokio_serde::formats::SymmetricalJson;

#[derive(Serialize, Deserialize, Clone)]
struct GreetRequest { name: String }

#[tarpc::service]
trait Greeter {
    async fn greet(req: GreetRequest) -> String;
}

#[derive(Clone)] struct GreeterServer;
impl Greeter for GreeterServer {
    async fn greet(self, _: context::Context, req: GreetRequest) -> String {
        format!("Hello, {}!", req.name)
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Start TCP listener on a random port
    let mut listener = tcp::listen("127.0.0.1:0", SymmetricalJson::default).await?;
    let addr = listener.local_addr();
    println!("Listening on {addr}");

    // Spawn server
    tokio::spawn(async move {
        if let Some(Ok(transport)) = listener.next().await {
            BaseChannel::with_defaults(transport)
                .execute(GreeterServer.serve())
                .for_each(|f| async move { tokio::spawn(f); })
                .await;
        }
    });

    // Connect client
    let transport = tcp::connect(addr, SymmetricalJson::default).await?;
    let client = GreeterClient::new(client::Config::default(), transport).spawn();

    let reply = client.greet(context::current(), GreetRequest { name: "Bob".into() }).await?;
    println!("{reply}"); // Hello, Bob!

    Ok(())
}
```
```

--------------------------------

### Improved Server Diagnostics with #[tarpc::server]

Source: https://github.com/google/tarpc/blob/main/RELEASES.md

When using `#[tarpc::server]`, `async fn`s are rewritten. This proc macro now provides better diagnostics for cases where non-async functions are used, helping to identify missing associated types.

```rust
error: not all trait items implemented, missing: `HelloFut`
 --> $DIR/tarpc_server_missing_async.rs:9:1
  |
9 | impl World for HelloServer {
| ^^^^

error: hint: `#[tarpc::server]` only rewrites async fns, and `fn hello` is not async
  --> $DIR/tarpc_server_missing_async.rs:10:5
   |
10 |     fn hello(name: String) ->  String {
|     ^^
```

--------------------------------

### Implement tarpc Service

Source: https://github.com/google/tarpc/blob/main/tarpc/README.md

Implement the generated service trait for your server struct. This contains the business logic for your RPCs.

```rust
// This is the type that implements the generated World trait. It is the business logic
// and is used to start the server.
#[derive(Clone)]
struct HelloServer;

impl World for HelloServer {
    async fn hello(self, _: context::Context, name: String) -> String {
        format!("Hello, {name}!")
    }
}
```

--------------------------------

### Implement Tarpc Service

Source: https://github.com/google/tarpc/blob/main/README.md

Implement the generated service trait for your server struct. This contains the business logic for your RPCs.

```rust
#[derive(Clone)]
struct HelloServer;

impl World for HelloServer {
    async fn hello(self, _: context::Context, name: String) -> String {
        format!("Hello, {name}!")
    }
}
```

--------------------------------

### Channel::execute

Source: https://context7.com/google/tarpc/llms.txt

High-level concurrent request dispatch. `Channel::execute(serve_fn)` is the simplest way to serve all requests on a channel concurrently. It returns a `Stream` of response futures that must be spawned or awaited.

```APIDOC
## `Channel::execute` — High-level concurrent request dispatch

`Channel::execute(serve_fn)` is the simplest way to serve all requests on a channel concurrently. It returns a `Stream` of response futures that must be spawned or awaited.

### Usage

- `BaseChannel::with_defaults(rx).execute(serve_fn)`: Executes the provided `serve_fn` for each incoming request, returning a stream of response futures.
- The stream of response futures should be handled (e.g., spawned or awaited) to process responses.
```

--------------------------------

### Client-side Round-Robin Load Balancing

Source: https://context7.com/google/tarpc/llms.txt

Implement client-side load balancing by wrapping multiple server stubs with `stub::RoundRobin`. This cycles through server connections atomically on each call for load distribution. Requires `futures::prelude` and `tarpc` client modules.

```rust
use tarpc::{
    client::{self, stub::load_balance::RoundRobin},
    context,
    server::{self, BaseChannel, Channel, serve},
    transport,
};
use futures::prelude::*;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Spin up two in-process servers
    let mut channels = Vec::new();
    for i in 0..2u32 {
        let (tx, rx) = transport::channel::unbounded();
        let server_id = i;
        tokio::spawn(
            BaseChannel::with_defaults(rx)
                .execute(serve(move |_ctx, n: u32| async move { Ok(n + server_id * 100) }))
                .for_each(|f| async move { tokio::spawn(f); }),
        );
        channels.push(client::new(client::Config::default(), tx).spawn());
    }

    // Build a round-robin stub over the two channels
    let lb = RoundRobin::new(channels);

    // Calls alternate between the two servers (0, 100, 0, 100, ...)
    use tarpc::client::stub::Stub as _;
    for _ in 0..4 {
        let resp = lb.call(context::current(), 1u32).await?;
        println!("{resp}"); // alternates: 1, 101, 1, 101
    }

    Ok(())
}
```

--------------------------------

### Add tarpc to Cargo.toml

Source: https://github.com/google/tarpc/blob/main/tarpc/README.md

To use tarpc, add the following line to your Cargo.toml file under [dependencies].

```toml
tarpc = "0.37"
```

--------------------------------

### Low-level server channel with request lifecycle management

Source: https://context7.com/google/tarpc/llms.txt

Use `server::BaseChannel` to wrap any `Transport` and manage request lifecycles, including deadline expiry and cancellation. It implements `Stream<Item = TrackedRequest>` and `Sink<Response>`.

```rust
use tarpc::{
    client,
    context,
    server::{self, BaseChannel, Channel, serve},
    transport,
};
use futures::prelude::*;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let (tx, rx) = transport::channel::unbounded();

    // BaseChannel::with_defaults uses server::Config { pending_response_buffer: 100 }
    let channel = BaseChannel::with_defaults(rx);

    // client::new returns NewClient { client, dispatch }; spawn the dispatch task
    let client = client::new(client::Config::default(), tx).spawn();

    tokio::spawn(async move {
        // Iterate manually over in-flight requests
        let mut requests = channel.requests();
        while let Some(Ok(request)) = requests.next().await {
            // execute() respects deadlines and cancellation automatically
            tokio::spawn(request.execute(serve(|_ctx, i: i32| async move { Ok(i * 2) })));
        }
    });

    let result = client.call(context::current(), 21i32).await?;
    println!("{result}"); // 42

    Ok(())
}
```

--------------------------------

### RoundRobin Client Stub

Source: https://context7.com/google/tarpc/llms.txt

Implements client-side round-robin load balancing by cycling through multiple server stubs. This allows for distributing requests across different server connections.

```APIDOC
## `stub::RoundRobin` — Client-side round-robin load balancing

`RoundRobin<Stub>` wraps multiple `Stub` implementations and cycles through them atomically on each call, enabling load distribution across multiple server connections.

### Usage Example
```rust
use tarpc::{
    client::{self, stub::load_balance::RoundRobin},
    context,
    server::{self, BaseChannel, Channel, serve},
    transport,
};
use futures::prelude::*;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Spin up two in-process servers
    let mut channels = Vec::new();
    for i in 0..2u32 {
        let (tx, rx) = transport::channel::unbounded();
        let server_id = i;
        tokio::spawn(
            BaseChannel::with_defaults(rx)
                .execute(serve(move |_ctx, n: u32| async move { Ok(n + server_id * 100) }))
                .for_each(|f| async move { tokio::spawn(f); }),
        );
        channels.push(client::new(client::Config::default(), tx).spawn());
    }

    // Build a round-robin stub over the two channels
    let lb = RoundRobin::new(channels);

    // Calls alternate between the two servers (0, 100, 0, 100, ...)
    use tarpc::client::stub::Stub as _;
    for _ in 0..4 {
        let resp = lb.call(context::current(), 1u32).await?;
        println!("{resp}"); // alternates: 1, 101, 1, 101
    }

    Ok(())
}
```
```

--------------------------------

### Implement Request Hooks for Server Middleware in Rust

Source: https://context7.com/google/tarpc/llms.txt

Use `RequestHook` trait's `before`, `after`, and `before_and_after` methods to add middleware logic to tarpc server handlers. This allows for request validation, logging, and response manipulation.

```rust
use futures::executor::block_on;
use tarpc::{
    ServerError,
    context,
    server::{serve, request_hook::{BeforeRequest, AfterRequest, RequestHook}},
};
use std::{io, time::Instant};

// --- before hook: reject requests with value 0 ---
let handler = serve(|_ctx, n: i32| async move { Ok(n + 1) })
    .before(|_ctx: &mut context::Context, req: &i32| {
        let req = *req;
        async move {
            if req == 0 {
                Err(ServerError::new(io::ErrorKind::InvalidInput, "zero not allowed".into()))
            } else {
                Ok(())
            }
        }
    });

assert!(block_on(handler.serve(context::current(), 0)).is_err());
assert_eq!(block_on(handler.serve(context::current(), 4)).unwrap(), 5);

// --- before_and_after hook: measure latency ---
struct Latency(Instant);

impl<Req> BeforeRequest<Req> for Latency {
    async fn before(&mut self, _ctx: &mut context::Context, _req: &Req) -> Result<(), ServerError> {
        self.0 = Instant::now();
        Ok(())
    }
}
impl<Resp> AfterRequest<Resp> for Latency {
    async fn after(&mut self, _ctx: &mut context::Context, _resp: &mut Result<Resp, ServerError>) {
        tracing::info!(elapsed_ms = self.0.elapsed().as_millis(), "request complete");
    }
}

let tracked = serve(|_ctx, n: u64| async move { Ok(n * 2) })
    .before_and_after(Latency(Instant::now()));
assert_eq!(block_on(tracked.serve(context::current(), 21u64)).unwrap(), 42);
```

--------------------------------

### server::BaseChannel

Source: https://context7.com/google/tarpc/llms.txt

Low-level server channel with request lifecycle management. It wraps any `Transport` and implements `Stream<Item = TrackedRequest>` + `Sink<Response>`, handling deadline expiry, duplicate request detection, and cascading cancellation automatically.

```APIDOC
## `server::BaseChannel` — Low-level server channel with request lifecycle management

`BaseChannel` is the foundational server type. It wraps any `Transport` and implements `Stream<Item = TrackedRequest>` + `Sink<Response>`. It handles deadline expiry, duplicate request detection, and cascading cancellation automatically.

### Constructor

- **BaseChannel::with_defaults(rx)**: Creates a new `BaseChannel` with default server configuration (e.g., `server::Config { pending_response_buffer: 100 }`).
```

--------------------------------

### TCP Network Transport with Serde in Rust

Source: https://context7.com/google/tarpc/llms.txt

Utilize `serde_transport::tcp` for network communication over TCP, integrating Serde for JSON serialization. This requires enabling the `serde-transport` and `tcp` features.

```toml
# Cargo.toml
tarpc = { version = "0.37", features = ["full"] }
tokio-serde = { version = "0.9", features = ["json"] }
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["full"] }
```

```rust
use futures::prelude::*;
use serde::{Deserialize, Serialize};
use tarpc::{
    client,
    context,
    server::{self, BaseChannel, Channel},
    serde_transport::tcp,
};
use tokio_serde::formats::SymmetricalJson;

#[derive(Serialize, Deserialize, Clone)]
struct GreetRequest { name: String }

#[tarpc::service]
trait Greeter {
    async fn greet(req: GreetRequest) -> String;
}

#[derive(Clone)] struct GreeterServer;
impl Greeter for GreeterServer {
    async fn greet(self, _: context::Context, req: GreetRequest) -> String {
        format!("Hello, {}!", req.name)
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Start TCP listener on a random port
    let mut listener = tcp::listen("127.0.0.1:0", SymmetricalJson::default).await?;
    let addr = listener.local_addr();
    println!("Listening on {addr}");

    // Spawn server
    tokio::spawn(async move {
        if let Some(Ok(transport)) = listener.next().await {
            BaseChannel::with_defaults(transport)
                .execute(GreeterServer.serve())
                .for_each(|f| async move { tokio::spawn(f); })
                .await;
        }
    });

    // Connect client
    let transport = tcp::connect(addr, SymmetricalJson::default).await?;
    let client = GreeterClient::new(client::Config::default(), transport).spawn();

    let reply = client.greet(context::current(), GreetRequest { name: "Bob".into() }).await?;
    println זיי{reply}" זיי; // Hello, Bob!

    Ok(())
}
```

--------------------------------

### Automatic Request Retry on Failure

Source: https://context7.com/google/tarpc/llms.txt

Wrap any stub with `stub::Retry` to automatically retry requests based on a provided predicate. The predicate `F: Fn(&Result<Resp, RpcError>, attempt) -> bool` determines if a retry should occur. Requests are wrapped in `Arc` for efficient cloning.

```rust
use std::sync::{Arc, atomic::{AtomicU32, Ordering}};
use tarpc::{
    client::{RpcError, stub::{Stub, retry::Retry}},
    context,
    server::{self, BaseChannel, Channel, serve},
    transport,
};
use futures::prelude::*;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let attempt_counter = Arc::new(AtomicU32::new(0));
    let counter_clone = attempt_counter.clone();

    let (tx, rx) = transport::channel::unbounded();
    tokio::spawn(
        BaseChannel::with_defaults(rx)
            .execute(serve(move |_ctx, n: u32| {
                let count = counter_clone.fetch_add(1, Ordering::SeqCst);
                async move {
                    // Fail the first two attempts
                    if count < 2 {
                        Err(tarpc::ServerError::new(std::io::ErrorKind::Unavailable, "busy".into()))
                    } else {
                        Ok(n * 10)
                    }
                }
            }))
            .for_each(|f| async move { tokio::spawn(f); }),
    );

    let base_stub = client::new(client::Config::default(), tx).spawn();
    // Wrap in Arc because Retry requires Stub<Req = Arc<Req>>
    // Retry up to 5 times on any error
    let retry_stub = Retry::new(base_stub, |result: &Result<u32, RpcError>, attempt: u32| {
        result.is_err() && attempt < 5
    });

    let resp = retry_stub.call(context::current(), 7u32).await?;
    println!("{resp}"); // 70 (succeeds on 3rd attempt)
    println!("total attempts: {}", attempt_counter.load(Ordering::SeqCst)); // 3

    Ok(())
}
```

--------------------------------

### Unix Domain Socket Transport

Source: https://context7.com/google/tarpc/llms.txt

Connects to and listens on Unix Domain Sockets for low-latency local IPC. This is an alternative to TCP transport, offering similar API but optimized for local communication.

```APIDOC
## `serde_transport::unix::connect` / `listen` — Unix Domain Socket transport

Requires features `serde-transport` and `unix` (Linux/macOS). Identical API to TCP but uses Unix Domain Sockets for lower-latency local IPC.

### Usage Example
```rust
use futures::prelude::*;
use tarpc::{
    client,
    context,
    server::{self, BaseChannel, Channel, serve},
    serde_transport::unix,
};
use tokio_serde::formats::SymmetricalJson;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // TempPathBuf creates a socket file and removes it on drop
    let sock = unix::TempPathBuf::with_random("my_service");

    let mut listener = unix::listen(&sock, SymmetricalJson::<i32>::default).await?;
    tokio::spawn(async move {
        if let Some(Ok(transport)) = listener.next().await {
            BaseChannel::with_defaults(transport)
                .execute(serve(|_ctx, n: i32| async move { Ok(n + 1) }))
                .for_each(|f| async move { tokio::spawn(f); })
                .await;
        }
    });

    let transport = unix::connect(&sock, SymmetricalJson::<i32>::default).await?;
    let client = client::new(client::Config::default(), transport).spawn();
    let result = client.call(context::current(), 41).await?;
    println!("{result}"); // 42

    Ok(())
}
```
```

--------------------------------

### Limit Concurrent Requests in tarpc Server

Source: https://context7.com/google/tarpc/llms.txt

Use `Channel::max_concurrent_requests` to cap the number of in-flight requests per connection. Exceeding the limit will result in a `ServerError`, protecting the server from overload.

```rust
use tarpc::{client, context, server::{self, BaseChannel, Channel, serve}, transport};
use futures::prelude::*;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let (tx, rx) = transport::channel::unbounded();
    let client = client::new(client::Config::default(), tx).spawn();

    tokio::spawn(
        // Allow at most 5 concurrent requests on this channel
        BaseChannel::with_defaults(rx)
            .max_concurrent_requests(5)
            .execute(serve(|_ctx, n: u32| async move { Ok(n) }))
            .for_each(|f| async move { tokio::spawn(f); }),
    );

    let resp = client.call(context::current(), 99u32).await?;
    println זיי{resp}" זיי; // 99

    Ok(())
}
```

--------------------------------

### Retry Client Stub

Source: https://context7.com/google/tarpc/llms.txt

Provides automatic request retry functionality on failure. It wraps a stub and retries requests based on a configurable predicate, useful for handling transient network issues or server unavailability.

```APIDOC
## `stub::Retry` — Automatic request retry on failure

`Retry<F, Stub>` wraps any `Stub` with a retry predicate `F: Fn(&Result<Resp, RpcError>, attempt) -> bool`. Requests are wrapped in `Arc` so they can be cheaply cloned across retries without requiring `Clone` on the request type.

### Usage Example
```rust
use std::sync::{Arc, atomic::{AtomicU32, Ordering}};
use tarpc::{
    client::{RpcError, stub::{Stub, retry::Retry}},
    context,
    server::{self, BaseChannel, Channel, serve},
    transport,
};
use futures::prelude::*;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let attempt_counter = Arc::new(AtomicU32::new(0));
    let counter_clone = attempt_counter.clone();

    let (tx, rx) = transport::channel::unbounded();
    tokio::spawn(
        BaseChannel::with_defaults(rx)
            .execute(serve(move |_ctx, n: u32| {
                let count = counter_clone.fetch_add(1, Ordering::SeqCst);
                async move {
                    // Fail the first two attempts
                    if count < 2 {
                        Err(tarpc::ServerError::new(std::io::ErrorKind::Unavailable, "busy".into()))
                    } else {
                        Ok(n * 10)
                    }
                }
            }))
            .for_each(|f| async move { tokio::spawn(f); }),
    );

    let base_stub = client::new(client::Config::default(), tx).spawn();
    // Wrap in Arc because Retry requires Stub<Req = Arc<Req>>
    // Retry up to 5 times on any error
    let retry_stub = Retry::new(base_stub, |result: &Result<u32, RpcError>, attempt: u32| {
        result.is_err() && attempt < 5
    });

    let resp = retry_stub.call(context::current(), 7u32).await?;
    println!("{resp}"); // 70 (succeeds on 3rd attempt)
    println!("total attempts: {}", attempt_counter.load(Ordering::SeqCst)); // 3

    Ok(())
}
```
```

--------------------------------

### RequestHook Middleware

Source: https://context7.com/google/tarpc/llms.txt

The `RequestHook` trait allows for composable middleware on server handlers. It supports `before`, `after`, and `before_and_after` hooks for request validation, logging, and response mutation.

```APIDOC
## `RequestHook::before` / `after` / `before_and_after` — Middleware for server handlers

The `RequestHook` trait (auto-implemented for all `Serve` types) enables composable middleware. Hooks can validate requests, enforce deadlines, log latency, or mutate responses — chained in a type-safe, zero-allocation manner.

### Example: Rejecting requests with value 0 using `before` hook
```rust
use futures::executor::block_on;
use tarpc::{ServerError, context, server::{serve, request_hook::{BeforeRequest, AfterRequest, RequestHook}}};
use std::{io, time::Instant};

let handler = serve(|_ctx, n: i32| async move { Ok(n + 1) })
    .before(|_ctx: &mut context::Context, req: &i32| {
        let req = *req;
        async move {
            if req == 0 {
                Err(ServerError::new(io::ErrorKind::InvalidInput, "zero not allowed".into()))
            } else {
                Ok(())
            }
        }
    });

assert!(block_on(handler.serve(context::current(), 0)).is_err());
assert_eq!(block_on(handler.serve(context::current(), 4)).unwrap(), 5);
```

### Example: Measuring latency using `before_and_after` hook
```rust
struct Latency(Instant);

impl<Req> BeforeRequest<Req> for Latency {
    async fn before(&mut self, _ctx: &mut context::Context, _req: &Req) -> Result<(), ServerError> {
        self.0 = Instant::now();
        Ok(())
    }
}
impl<Resp> AfterRequest<Resp> for Latency {
    async fn after(&mut self, _ctx: &mut context::Context, _resp: &mut Result<Resp, ServerError>) {
        tracing::info!(elapsed_ms = self.0.elapsed().as_millis(), "request complete");
    }
}

let tracked = serve(|_ctx, n: u64| async move { Ok(n * 2) })
    .before_and_after(Latency(Instant::now()));
assert_eq!(block_on(tracked.serve(context::current(), 21u64)).unwrap(), 42);
```
```

--------------------------------

### Configure TCP Transport Max Frame Length

Source: https://github.com/google/tarpc/blob/main/RELEASES.md

Configure the maximum frame length for TCP transport connections. This is useful for sending requests or responses larger than the default payload limit.

```rust
let mut transport = tarpc::serde_transport::tcp::connect(server_addr, Json::default);
transport.config_mut().max_frame_length(4294967296);
let mut client = MyClient::new(client::Config::default(), transport.await?).spawn()?;

```

--------------------------------

### Channel::max_concurrent_requests

Source: https://context7.com/google/tarpc/llms.txt

Limits the number of in-flight requests per connection to prevent overwhelming the server. Returns a `ServerError` if the limit is exceeded.

```APIDOC
## `Channel::max_concurrent_requests` — Limit in-flight requests per connection

Wraps a `Channel` to return a `ServerError` once the concurrent request limit is exceeded, protecting the server from being overwhelmed.

### Example: Limiting concurrent requests to 5
```rust
use tarpc::{client, context, server::{self, BaseChannel, Channel, serve}, transport};
use futures::prelude::*;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let (tx, rx) = transport::channel::unbounded();
    let client = client::new(client::Config::default(), tx).spawn();

    tokio::spawn(
        // Allow at most 5 concurrent requests on this channel
        BaseChannel::with_defaults(rx)
            .max_concurrent_requests(5)
            .execute(serve(|_ctx, n: u32| async move { Ok(n) }))
            .for_each(|f| async move { tokio::spawn(f); })
    );

    let resp = client.call(context::current(), 99u32).await?;
    println!("{resp}"); // 99

    Ok(())
}
```
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.