---
description: How to build a service with RustStream (handlers, app, brokers, codecs)
globs: examples/**/*.rs
alwaysApply: false
---
- A handler is an `async fn` annotated with `#[subscriber("subject")]`. Its first parameter is the
decoded payload by reference (`&T`, where `T: serde::Deserialize`). Further parameters are
extractors: `State<T>` injects a component of the application state, `Ctx<K>` injects one broker
context field (partition, offset, ...). An explicit `ctx: &mut Context` parameter is optional and
last; omit it unless you use it.
- The return type drives acknowledgement: `HandlerResult`, `()` (always acks), or `Result<_, E>` (an
`Err` nacks). For a reply handler, add a `publish("subject")` clause and return the reply value,
or `Result<Reply, HandlerResult>` for explicit ack control.
- Build the service with `#[ruststream::app]` on a synchronous `fn app() -> RustStream` (or
`impl App` when layers/state change the type). Do not write your own `main` or `#[tokio::main]`.
- Application state is built once in `on_startup` (async closures: `async move |()| ...`) and
injected with `State<T>` via `#[derive(FromRef)]` on the state struct. There is no runtime
type-map; everything is typed and zero-cost.
- Construct brokers synchronously (`MemoryBroker::new()`, `NatsBroker::new(url)`); the runtime
connects them at startup. Never connect inside the builder.
- The codec defaults to `codec::DefaultCodec` (`json` > `cbor` > `msgpack`), so `include` needs no
codec argument. Log with `tracing`, never `println!`. In tests use `MemoryBroker` driven by the
`TestApp` harness (`testing` feature).
## Subscriber and app
The macro turns `handle` into a value named after the function; mount it with `include`.
```rust
use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, HandlerResult, RustStream};
use ruststream::subscriber;
use serde::Deserialize;
#[derive(Debug, Deserialize)]
struct Order {
id: u64,
}
#[subscriber("orders.created")]
async fn handle(order: &Order) -> HandlerResult {
tracing::info!(order.id, "processing order");
HandlerResult::Ack
}
#[ruststream::app]
fn app() -> RustStream {
RustStream::new(AppInfo::new("orders", "0.1.0"))
.with_broker(MemoryBroker::new(), |b| {
b.include(handle);
})
}
```
Swap the broker without touching the handlers (NATS connects lazily at startup):
```rust
use ruststream_nats::NatsBroker;
.with_broker(NatsBroker::new("nats://localhost:4222"), |b| b.include(handle))
```
## Handler results
```rust
// explicit ack / nack
HandlerResult::Ack
HandlerResult::Nack { requeue: true }
// `()` always acks; `Result<(), E>` acks on Ok, nacks on Err
async fn handle(order: &Order) -> Result<(), MyError> { Ok(()) }
```
## Replying (request/response)
Return the reply value and add `publish("subject")`; hand `include_publishing` a `TypedPublisher`.
```rust
use ruststream::runtime::TypedPublisher;
use serde::Serialize;
#[derive(Serialize)]
struct Confirmation {
id: u64,
accepted: bool,
}
#[subscriber("orders", publish("confirmations"))]
async fn confirm(order: &Order) -> Confirmation {
Confirmation { id: order.id, accepted: true }
}
// inside with_broker(broker, |b| { ... }):
let replies = TypedPublisher::new(b.broker().publisher()); // default codec
b.include_publishing(confirm, replies);
```
Return `Result<Reply, HandlerResult>` to control acknowledgement: `Ok(reply)` publishes and acks,
`Err(result)` publishes nothing and the dispatcher acts on the returned `HandlerResult`. Spell the
`Result` out in the signature (a type alias is treated as a plain reply type). A failed reply
publish nacks the incoming message with `requeue = true`.
Note the `publish(..)` clause takes a string literal; runtime-computed destinations are not
supported there.
## State injection (dependency injection)
Build the state once in `on_startup`; `#[derive(FromRef)]` makes every field injectable with
`State<FieldType>` - no extractor impls, no type-map.
```rust
use std::convert::Infallible;
use ruststream::FromRef;
use ruststream::runtime::{AppInfo, HandlerResult, RustStream, State};
use ruststream::subscriber;
#[derive(Clone)]
struct Database; // connection pool, etc.
#[derive(FromRef)]
struct AppState {
db: Database,
}
#[subscriber("orders")]
async fn handle(order: &Order, State(db): State<Database>) -> HandlerResult {
// db.store(order).await;
let _ = db;
HandlerResult::Ack
}
#[ruststream::app]
fn app() -> impl ruststream::runtime::App {
RustStream::new(AppInfo::new("orders", "0.1.0"))
.on_startup(async move |()| Ok::<_, Infallible>(AppState { db: Database }))
.with_broker(MemoryBroker::new(), |b| b.include(handle))
}
```
## Broker context fields (`Ctx<K>`)
`Ctx<K>` injects one field of the broker's per-delivery context, the way `State<T>` injects a
state component. The key `K` is a broker-provided `ContextField` type; a handler using only `Ctx`
extractors needs no `&mut Context` parameter at all.
```rust
use ruststream::runtime::Ctx;
// Keys come from the broker crate, e.g. ruststream_rdkafka::context::{Partition, Offset}.
#[subscriber("orders")]
async fn handle(order: &Order, Ctx(partition): Ctx<Partition>) -> HandlerResult {
tracing::info!(partition, "delivery");
HandlerResult::Ack
}
```
## Context (optional explicit parameter)
When you need the raw delivery objects, take `ctx: &mut Context<'_, C, S>` as the last parameter
(`C` = broker context type, `S` = application state):
- `ctx.name()` - the subject/channel the message arrived on.
- `ctx.headers()` / `ctx.headers_mut()` - the working copy of the message headers.
- `ctx.state()` - the shared application state built by `on_startup`.
There is no `ctx.get::<T>()` and no named-publisher registry; state is typed (`State<T>` /
`ctx.state()`), and publishing goes through `publish(..)` replies or a `TypedPublisher` in state.
## Choosing a codec
`include` uses `DefaultCodec`. No mounting call takes a codec argument; change the default for a
scope with `with_broker_codec`, or for a `Router` chain with `with_codec`.
```rust
use ruststream::codec::CborCodec;
// every handler in this scope decodes with CborCodec
RustStream::new(info).with_broker_codec(broker, CborCodec, |b| {
b.include(handle);
});
// on a Router: with_codec applies to the includes after it (it can change mid-chain)
Router::<MemoryBroker>::new().with_codec(CborCodec).include(handle);
```
## Routers
Collect handlers in their own module, then mount the group. A `Router` is a consuming builder
(each call returns a new type), so the calls chain and the function returns `impl RouterDef<B>`.
The app's global `.layer(..)` reaches router handlers, so cross-cutting middleware (logging,
metrics) goes on the app and applies everywhere; that global layer must be a `BlanketLayer`
(every bundled layer is). Publish-side middleware (tracing propagation, broker envelopes such as
Kafka's Schema Registry framing) is added app-wide with `.publish_layer(..)` before `with_broker`.
```rust
use ruststream::runtime::{Router, RouterDef};
// `use<>` opts out of borrowing the broker (the router owns its Arc-backed publisher).
fn orders(broker: &MemoryBroker) -> impl RouterDef<MemoryBroker> + use<> {
let replies = TypedPublisher::new(broker.publisher());
Router::new()
.include(handle)
.include_publishing(confirm, replies)
}
// in main
.with_broker(MemoryBroker::new(), |b| b.include_router(orders(b.broker())))
```
## Broker-specific subscriptions
For options the by-name form can't express (NATS JetStream, Kafka consumer groups / start offsets),
pass the broker's descriptor directly in the attribute, or override the source with `include_on`.
The codec resolves the same way as for `include`.
```rust
use ruststream_nats::SubscribeOptions;
b.include_on(
SubscribeOptions::new("orders.*").jetstream("ORDERS").durable("worker"),
handle,
);
```
## AsyncAPI metadata
Decoding only needs `serde::Deserialize`. The extra derives are **documentation only** for the
generated AsyncAPI spec and are optional: `ruststream::Message` names/describes the type, and
`JsonSchema` (feature `asyncapi`) emits its payload schema.
Neither selects a wire codec - that stays the default (`json` > `cbor` > `msgpack`) or whatever
`with_codec` / `with_broker_codec` names. `JsonSchema` describes the payload's logical shape in the
JSON Schema language (as in OpenAPI), independent of how bytes are encoded on the wire; you cannot
attach a codec to a type.
```rust
use ruststream::Message;
use ruststream::schemars::JsonSchema;
use serde::Deserialize;
/// An order placed on the orders channel.
#[derive(Debug, Deserialize, Message, JsonSchema)]
struct Order {
id: u64,
}
```
## Testing
Use the `TestApp` harness (`testing` feature): it drives the built application with no network and
no server, injects input through the broker bus, and exposes per-broker assertions. `MemoryBroker`
is a real broker here, driven through the same dispatch path as production.
```rust
use ruststream::memory::MemoryBroker;
use ruststream::testing::TestApp;
let tb = TestApp::start(service()).await?;
tb.broker::<MemoryBroker>()
.publish("orders", &Order { id: 42 })
.await?;
tb.broker::<MemoryBroker>()
.subscriber("orders")
.assert_called_once()
.with(&Order { id: 42 })
.settled(HandlerResult::Ack);
tb.broker::<MemoryBroker>()
.published::<Receipt>("receipts")
.assert_called_once()
.with(&Receipt { order_id: 42 });
```
Broker crates ship their own in-process test broker under their `testing` feature; `TestApp` drives
those the same way.