---
description: How to build a service with RustStream (handlers, app, brokers, codecs)
globs: examples/**/*.rs
alwaysApply: false
---
- A handler body imports one glob: `use ruststream::prelude::*;`. It carries the application
builder, the handler surface (`HandlerOutcome`, `Context`), every extractor (`State`, `Ctx`,
`Headers`, `Out`), the mount-site settings, the publish builder (`PublishExt`), the publisher
capability traits a slot bound names (`Publisher`, `TransactionalPublisher`, `OwnedTransactions`,
`RequestReply`), and the attribute macros and derives. A body never names a publish policy, so it
never knows which broker runs it. The one exception is a body that adjusts a per-message broker
setting: it imports that broker's prelude for the step and names its options type in the bound.
- A mount site imports the BROKER's glob (`use ruststream_nats::prelude::*;`,
`use ruststream::memory::prelude::*;`): it re-exports the core prelude, the broker, its
subscription descriptor and its context keys, plus its publish policies under the uniform names
`Publish`, `TransactionalPublish` and `Request`, so the chain reads the same whichever broker it
is on and swapping brokers swaps the glob. A file holding both a body and its mount site is
served by the broker glob alone. Two broker preludes can be globbed side by side - the core
items they share resolve to the same types.
- Only what a given service actually reaches for stays an explicit import: codecs
(`ruststream::codec`) and the optional feature modules (`ruststream::testing`, `asyncapi`,
`metrics`, `logging`, `otel`). `Router` / `RouterDef`, `Reply`, `DefaultSlot`, `OutSlot`,
`FailurePolicies` / `FailurePolicy`, `Workers`, `Unnamed`, `nonzero!`, the transform vocabulary
(`PublishTransform`, `ForReply` / `ForSlot`, `Reads` / `Names`) and the manual path's `Handle` /
`subscriber` pair are all in the prelude.
- A handler is an `async fn` annotated with `#[subscriber("subject")]`. Its first parameter is the
message, and its type alone picks the form: `&T` is one message and `&[T]` a batch, while the
lane is the type's own business - `T: serde::Deserialize` decodes with the codec, a
`#[derive(Deserialized)]` type builds itself from the raw bytes (`&Frame<'_>`, batch
`&[Frame<'_>]`), and a type a broker crate's derive puts on the carried lane is the value the
delivery already holds (a row a database driver read; a batch lends them as one slice). No
clause names any of this. Further parameters are extractors:
`State<T>` injects a component of the application state, `Ctx<K>` one broker context field,
`Headers<T>` the delivery headers parsed into a typed contract, and any type implementing
`FromContext` is one you wrote yourself (an auth guard, a request-scoped lookup); an extractor
that rejects settles the delivery and the body never runs. `Out<impl Publisher, ..>` is not an
extractor but an injection: a live publisher the mount site's policy built. An explicit
`ctx: &mut Context` parameter is optional and comes right after the payload; omit it unless you
use it.
- The return type drives acknowledgement: `HandlerOutcome`, `()` (always acks), `Result<_, E>` (an
`Err` drops - a nack with no requeue), `Result<HandlerOutcome, E>`, or `Vec<HandlerOutcome>` on a
batch to settle entry by entry. For a reply handler, add a `reply` clause and return the reply
value, or `Result<Reply, HandlerOutcome>` for explicit ack control (`Vec<Reply>` and
`Result<Vec<Reply>, HandlerOutcome>` are the batch counterparts).
- Where a reply goes is a property of the reply type, not of the mount site. Every reply type
derives `Outgoing`, and a type without it is a compile error at the handler naming the derive.
`#[outgoing(name = "dest")]` on the type is the destination and the clause stays bare
(`reply`); a derive naming nothing takes the clause's name (`reply("dest")`). A declared name
wins over a mount-site one, and startup logs a warning when the two differ. A name template
does not resolve on the reply path - nothing binds its placeholders there. The generated AsyncAPI
document reports the resolved destination, so a declaration cannot document one channel and
publish to another.
- 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.
- `include` is the only mounting call. It takes the definition and returns a mount chain:
`.out_reply(policy)` names the reply's publish policy, `.out(<marker>, policy)` one injected
publisher's (finish a slot chain with `.build()`), `.out_retry(policy)` replaces the publisher a
`retry_after` copy leaves through, and `max_attempts(n)` / `dead_letter(name)` cap the
registration's retries; chain nothing and the broker's default publish policy is used. A definition never names a publisher: it declares what a handler replies with and where,
and a policy belongs to a broker, so it is named where a broker is. The chain is a statement:
`with_broker(broker, |b| { b.include(handle); })`, not `|b| b.include(handle)`.
- The codec defaults to `codec::DefaultCodec` (`json` > `cbor` > `msgpack`), so `include` needs no
codec argument. With no codec feature at all, anything that would need to encode or decode is a
compile error naming the fix; the `Deserialized`/`Serialized` byte lanes never need a codec.
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::prelude::*;
use serde::Deserialize;
#[derive(Debug, Deserialize)]
struct Order {
id: u64,
}
#[subscriber("orders.created")]
async fn handle(order: &Order) -> HandlerOutcome {
tracing::info!(order.id, "processing order");
HandlerOutcome::ack()
}
#[ruststream::app]
fn app() -> RustStream {
RustStream::new(AppInfo::new("orders", "0.1.0"))
.with_broker(MemoryBroker::new(), |b| {
b.include(handle);
})
}
```
Swapping the broker is swapping the glob; the handlers do not change (NATS connects at startup):
```rust
use ruststream_nats::prelude::*; // brings NatsBroker + the core prelude
.with_broker(NatsBroker::new("nats://localhost:4222"), |b| { b.include(handle); })
```
## Handler outcomes
Every settlement is one of four constructors: a handler then reads the same whether it acks or
nacks, and every outcome chains into `and_after`.
```rust
HandlerOutcome::ack() // ack
HandlerOutcome::retry() // nack, requeue
HandlerOutcome::drop() // nack, no requeue
HandlerOutcome::retry_after(delay) // nack with a delay - not ready yet, do not spin
// `()` always acks; `Result<(), E>` acks on Ok, nacks on Err
async fn handle(order: &Order) -> Result<(), MyError> { Ok(()) }
```
The broker status inside is not a public enum: read an outcome with the predicates `is_ack()`,
`is_retry()`, `is_drop()` and `retry_delay()`. On a broker without native delayed redelivery the
runtime honours `retry_after` by publishing a copy after the delay. Every registration whose copies
the runtime publishes already has a publisher for them, from the broker's default publish policy;
`.out_retry(policy)` (`.out(Retry, policy)` spelled out) replaces it. The position is an `Out`
slot, so `.codec(..)` and `.transform(..)` after it apply to the copy - which carries the delivery's
own bytes, so nothing re-encodes it.
A registration caps its retries at the mount site, right after `include` and before
`out_retry(..)`: `b.include(reconcile).max_attempts(nonzero!(5)).dead_letter("orders.dead")`.
Where the subscription's descriptor addresses no copies (a wildcard, an MQTT filter, a pattern),
`.out_retry(policy).to(name)` names where they go, or a transform on the position names one per
delivery; a registration that names neither is a compile error on a `Router` chain and refuses to
start on a scope. Each step is declared once. Below the cap a copy goes back to the subscription; at the cap it goes to
the dead-letter destination, or the delivery is rejected where none is named. A destination
declared without a cap takes over every copy. The count is the broker's own where the transport
keeps one (`IncomingMessage::redelivery_count`) and the framework's `x-ruststream-retry-count`
header otherwise, never both, and an immediate `retry()` obeys the same cap. A broker with native
delayed redelivery gets the delay only below the cap: the count is read first, and a delivery at
the cap is carried away instead of coming back - on a transport that counts its own deliveries and
on one whose crate honours the delay by publishing a copy that carries the header. The generated
`AsyncAPI` document reports the dead-letter destination as a channel the registration publishes
to.
## Post-settle work (`and_after`)
`and_after` attaches a continuation that runs **after** the message is settled and returns the
`HandlerOutcome` it was called on (so the handler's return type stays `HandlerOutcome`, or
`Vec<HandlerOutcome>` on a batch). Use it for a side effect that must not gate the settlement
decision or affect redelivery: a notification, slow follow-up work, a cache warm-up.
```rust
use ruststream::prelude::*;
#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
let id = order.id;
HandlerOutcome::ack().and_after(async move {
tracing::info!(id, "acked; notifying downstream");
})
}
// per element on a batch: the continuation rides with its own message
#[subscriber("orders")]
async fn handle_batch(orders: &[Order]) -> Vec<HandlerOutcome> {
orders
.iter()
.map(|order| {
if order.id == 0 {
HandlerOutcome::retry() // a plain outcome, no continuation
} else {
let id = order.id;
HandlerOutcome::ack().and_after(async move { follow_up(id).await })
}
})
.collect()
}
```
It chains off any outcome, `drop().and_after(..)` included. The continuation is **at-most-once**:
the message is already settled when it runs, so a panic or a crash never redelivers it. Work whose
loss must redeliver the message does not go here - settle by outcome and let the broker retry.
Graceful shutdown drains continuations still in flight. A continuation runs where the delivery ran:
on a `threads(n)` subscription, on the delivery's thread.
## Settings the mount site fills in
A `#[subscriber]` with no source in the attribute leaves the name to the mount; the definition
value carries a builder for the declarative settings, applied before `include` takes it.
```rust
use std::time::Duration;
// FailurePolicies, FailurePolicy and nonzero! ride the prelude; no second import
use ruststream::memory::prelude::*;
#[subscriber]
async fn bill(order: &Order) -> HandlerOutcome {
HandlerOutcome::ack()
}
// in with_broker(broker, |b| { ... }):
b.include(
bill.name("orders")
.workers(nonzero!(4))
.on_failure(FailurePolicies::default().with_decode(FailurePolicy::Skip)),
);
// a batch body: `batch(n)` is the batch size, the one batch parameter the framework carries, and it
// is mandatory - a batch handler mounted without it does not compile. The size goes to the broker,
// which builds its batches to it; the body sees exactly what the broker delivered. It applies to
// every batch shape and composes with a start position
b.include(settle.name("orders").batch(nonzero!(64)));
b.include(
drain
.name("orders")
.batch(nonzero!(128))
.start_at(MemoryPosition::start()),
);
// a broker's own batching options ride its subscription source, after the core's size:
// b.include(settle.name("orders").batch(nonzero!(6)).block(Duration::from_secs(5)));
// where the subscription opens, on a broker with a replayable log (Seekable)
b.include(audit.name("orders").start_at(MemoryPosition::start()));
```
`workers(n)`, `threads(n)`, `on_failure(..)` and `start_at(..)` can also be written as attribute
clauses (`#[subscriber("orders", workers(16, by_key))]`, `#[subscriber("images", threads(8))]`,
`#[subscriber("orders", on_failure(decode = retry))]`,
`#[subscriber("audit", start_at(MemoryPosition::start()))]`); those five (`reply` is the fifth)
are the only clauses, and `batch(n)` is mount-site only. `workers(n)` is concurrency for I/O on
the app's runtime; a handler that computes takes `threads(n)` (or `threads(n, by_key)`): `n`
dedicated threads of the subscription's own, each a current-thread runtime, where the handler
runs from the first poll to the end. The two fill one position, so a subscription names one.
What a `threads(n)` delivery leaves behind (a `tokio::spawn`, an `and_after` continuation, an
`after(..)` hook, a `retry_after` timer) stays on its thread; work goes to the app's runtime only
through `main.spawn(fut)` or `main.run(fut).await` on `Ctx(main): Ctx<MainRuntime>` (or
`ctx.main_runtime()`). A service with
`threads(n)` sizes the app's runtime for its I/O: `#[ruststream::app(flavor = "current_thread")]`
or `#[ruststream::app(worker_threads = 2)]`. The `poll-diagnostics` feature finds the handlers
that belong on `threads(n)`: every subscription samples its handler's time in `poll` and warns
when the average crosses the threshold; `.poll_diagnostics(PollDiagnostics::new().threshold(..))`
on `RustStream`, before `with_broker`, tunes it, and `metrics.observe_poll_diagnostics(&d)` or
`otel.observe_poll_diagnostics(&d)` exports it. `start_at` forces the position on
every startup and takes a value of the broker's own position type; without it the subscription
opens at the broker's default. A conditional default (apply only when no stored cursor exists for
the group) belongs on the broker's own subscription descriptor.
Seeking needs a log to seek in. `MemoryBroker::new()` keeps nothing, so the two forms are different
types: `MemoryBroker::retaining(Retention::Messages(nonzero!(64)))` is the one whose subscriptions
are `Seekable`, and a `start_at(..)` or a `SeekHandle` read only compiles against it. A handler
repositions its own subscription through the broker's context keys
(`Ctx(seeker): Ctx<SeekHandle>`, then `seeker.seek(MemoryPosition::sequence(3))`); a position the
retention bound already evicted reports `MemoryError::PositionEvicted`.
## Replying (request/response)
Return the reply value and write `reply` on the subscriber. The reply type derives `Outgoing`,
and its declaration decides where the answer goes: a type fixing its own name is published there
and the clause names nothing, a type naming nothing takes the clause's name. The reply publisher is
the broker's default policy unless the mount names one with `.out_reply(..)`.
```rust
use ruststream::memory::prelude::*;
use serde::Serialize;
// the destination on the type: the clause stays bare
#[derive(Debug, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
id: u64,
}
#[subscriber("receipt-requests", reply)]
async fn issue_receipt(order: &Order) -> Receipt {
Receipt { id: order.id }
}
// the destination left open: the mount site says
#[derive(Debug, Serialize, Outgoing)]
struct Confirmation {
id: u64,
accepted: bool,
}
#[subscriber("orders", reply("confirmations"))]
async fn confirm(order: &Order) -> Confirmation {
Confirmation { id: order.id, accepted: true }
}
// in with_broker(broker, |b| { ... }):
b.include(issue_receipt); // the type carries the destination
b.include(confirm); // broker's default publish policy
b.include(confirm).out_reply(Publish); // an explicit one
b.include(confirm).out_reply(Publish).codec(CborCodec).transform(Envelope); // the whole wiring
```
The steps after `.out_reply(..)` fill the reply wiring, each once: `.codec(..)` names the reply
codec (the default codec applies without it), `.transform(..)` composes a static
`PublishTransform`, `.batch_transform(..)` does the same on a batch's replies, and
`.transactional()` publishes a batch's replies inside one broker transaction. A second `.codec(..)`,
or any of these on a `Serialized` (byte-for-byte) reply, is a compile error.
Return `Result<Reply, HandlerOutcome>` to control acknowledgement: `Ok(reply)` publishes and acks,
`Err(outcome)` publishes nothing and the dispatcher acts on the returned `HandlerOutcome`. 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`.
The `reply(..)` clause takes a string literal, and a name template does not resolve here.
Answering somewhere else per delivery (a direct `reply-to`, a `ROUTER` peer identity) is a
transform; see [naming a destination from a transform](#naming-a-destination-from-a-transform).
## Raw payloads
When the payload is not a serialized value at all (a binary frame, a foreign wire format you parse
yourself), the payload type deserializes itself and the codec stays out of the path. The mnemonic
is in the trait names: `Deserialize` means the framework's codec builds the value, `Deserialized`
means the type builds itself. A scope codec does not apply, and this is the one subscriber form
available with no codec feature enabled at all. A bare `&[u8]` parameter is a compile error naming
the derive.
```rust
#[derive(Deserialized)]
struct Frame<'a>(&'a [u8]);
#[subscriber("frames")]
async fn ingest(frame: &Frame<'_>) -> HandlerOutcome {
parse(frame.0);
HandlerOutcome::ack()
}
// the form rule does not change with the lane - `&[Frame<'_>]` is the batch, the spelling
// comes with the derive, and the elements are borrowed from the batch's own messages for
// the duration of the call, so nothing is copied
#[subscriber("frames")]
async fn ingest_batch(frames: &[Frame<'_>]) -> HandlerOutcome {
HandlerOutcome::ack()
}
```
`on_failure(decode = ..)` applies here too: a constructor that validates (a flatbuffers root, a
length check) returns `Err` from `from_payload` and that policy settles the delivery, the same rung
a codec decode failure lands on. Extractors, `&mut Context`, `workers(..)`,
`on_failure(panic = ..)` and the injected `Out` parameters work unchanged on the single-delivery
shape; a batch of frames mounts in the plain form, with no reply and no `Out`.
There is one `reply("dest")` clause, and the reply *type* picks the wire: a `serde::Serialize`
reply encodes through the reply codec, a `#[derive(Serialized)]` newtype leaves byte-for-byte. The
two lanes are independent, so raw in / encoded out and decoded in / raw out (the gateway shape)
both compile. A `Vec<u8>` reply is not a byte reply - it encodes like any `Serialize` value - and a
`Serialized` batch reply does not exist.
```rust
#[derive(Outgoing, Serialized)]
struct Export(Vec<u8>);
#[subscriber("relay-in", reply("relay-out"))]
async fn relay(frame: &Frame<'_>) -> Export {
Export(frame.0.to_vec())
}
```
A `Serialized` reply names its publish policy at the include site the same way an encoded one does
(`b.include(relay).out_reply(Publish)`, or nothing for the broker's default); what it does not take
is the wiring steps after it, since its bytes leave with no codec and no transform stack.
A binary protocol (Protobuf, Cap'n Proto, FlatBuffers) is not a codec: the generated message *is*
its encoding, so it rides the same lanes and the model type stays visible at the mount site, in a
slot's dictionary and in the document. Three derives and one attribute put a whole schema there,
and a generator writes them for you (`prost_build`'s `message_attribute`):
```rust
#[derive(Clone, PartialEq, prost::Message, Outgoing, Serialized, Deserialized)]
#[outgoing(name = "orders")]
#[wire(prost)]
struct Order {
#[prost(uint64, tag = "1")]
id: u64,
}
```
`#[wire(prost)]` is shorthand for one generator's pair of functions; the general form names them,
`#[wire(encode = <path>, decode = <path>)]`, where `encode` is a `fn(&Self, &mut BytesMut)` and
`decode` a `fn(&[u8]) -> Result<Self, E>`. Neither lane resolves a codec, so such a service
compiles with no codec feature at all. For a custom serialization format you want *typed* handlers
for, implement `Codec` instead and keep the typed path.
## Publishing from a handler (`Out`)
The publisher arrives as a parameter. The mount binds each slot marker to a publish policy; the
runtime pairs it with the connected broker at startup, so the handler always holds a live
publisher - no registry, no erased lookup, no state plumbing.
```rust
use ruststream::memory::prelude::*;
use serde::Serialize;
#[derive(Debug, Serialize, Outgoing)]
struct Event {
id: u64,
}
#[subscriber("ingress")]
async fn forward(event: &Event, Out(out): Out<impl Publisher>) -> HandlerOutcome {
if out.message(event).to("egress").publish().await.is_err() {
return HandlerOutcome::retry();
}
HandlerOutcome::ack()
}
// in with_broker(broker, |b| { ... }): the single unnamed slot is DefaultSlot
b.include(forward).out(DefaultSlot, Publish).build();
```
### Several publishers: one marker per path
A handler that publishes down more than one path names a **slot marker** per parameter - a unit
struct deriving `OutSlot`, whose `#[publishes(..)]` list is what may leave through that path. The
include site binds each marker with `.out(marker, policy)` and commits with a terminal `.build()`;
the calls bind by marker, so their order does not matter. A forgotten binding is a compile error
naming the slot (`MissingSlot<Audit>`), never a runtime surprise. `DefaultSlot` (from
`ruststream::runtime`) names the implicit single slot when a handler also has a reply side.
```rust
// different paths, different message types
#[derive(OutSlot)]
#[publishes(OrderConfirmed, OrderPlaced)]
struct Orders;
#[derive(OutSlot)]
#[publishes(AuditEntry)]
struct Audit;
#[subscriber("orders.incoming")]
async fn route(
event: &Event,
Out(orders): Out<impl Publisher, Orders>,
Out(audit): Out<impl Publisher, Audit>,
) -> HandlerOutcome {
// publishing an AuditEntry through `orders` is a compile error: not in its list
HandlerOutcome::ack()
}
b.include(route)
.out(Audit, Publish)
.transform(OutboxEnvelope) // rides the slot the .out(..) before it bound
.out(Orders, Publish)
.build();
```
`.transform(..)` composes a `PublishTransform` onto the position named right before it, and onto
that one only; before any `.out(..)` the step has no position to ride. The step repeats, and the
transforms run in the order the chain writes them. On a registration that also replies,
`.transform(..)` after `.out_reply(..)` grows the reply's stack and after `.out(marker, ..)` the
slot's.
The capability in the bound can be refined - `Out<impl OwnedTransactions, Ledger>` compiles only
against a policy whose live publisher supports owned transactions, checked at the include site.
The bound names a broker capability trait (`Publisher`, `TransactionalPublisher`,
`OwnedTransactions`, `RequestReply`, or one a broker crate defines), never a broker type; on the
manual path the same bound sits on the entry's wired value,
`where L: OutEntry<Ledger, Wire: OwnedTransactions>`. Under it the entry offers that capability's
typed form - `message(..)`, `begin()`, `transaction()` - over the include site's codec and the
marker's list.
### Narrowing one path to a set of types
The optional **third** position declares what this handler sends through that slot. Omitted (or
`()`) it is unrestricted: the whole marker list, which is also what the generated document reports.
```rust
Out<impl Publisher, Orders> // whole #[publishes(..)] list
Out<impl Publisher, Orders, ()> // the same, spelled out
Out<impl Publisher, Orders, OrderConfirmed> // one type (it declares itself)
Out<impl Publisher, Orders, (OrderConfirmed, OrderPlaced)> // a tuple set
Out<impl Publisher, Orders, ConvertSends> // a reusable named set
```
A named set is a `#[derive(OutMessages)]` enum whose variants wrap the models. It is a type-level
declaration and is never constructed, so it is the way to share one set across handlers:
```rust
#[derive(OutMessages)]
enum ConvertSends {
Confirmed(OrderConfirmed),
Placed(OrderPlaced),
}
```
Publishing a type the set does not name is a compile error naming the missing membership. A type
that carries its own wire format belongs in the dictionary too, and the wire follows the type:
derive `Serialized` next to `#[derive(Outgoing)]`, list it in `#[publishes(..)]` like any model -
the document reports it under its own name with no payload schema - and publish it through the
same typed entry, `out.message(&export).publish()`: its bytes leave as they are, no codec runs,
and the destination comes off the declaration. Bytes with no declared type have no publish entry;
wrap them in a `Serialized` newtype.
### Where a message goes
The declaration on the message type decides which destination position the call site even has:
```rust
#[derive(Debug, Serialize, Outgoing)]
#[outgoing(name = "orders.confirmed")] // fixed
struct OrderConfirmed { id: u64 }
#[derive(Debug, Serialize, Outgoing)]
#[outgoing(name = "orders.{tenant}.placed")] // template
struct OrderPlaced { id: u64 }
#[derive(Debug, Serialize, Outgoing)] // no name: the call site says
struct Event { id: u64 }
#[derive(Debug, Outgoing, Serialized)] // bytes that already are the payload
struct Export(Vec<u8>);
```
```rust
// fixed name: resolved, so there is no `to(..)` to write
orders.message(&OrderConfirmed { id }).publish().await?;
// template: `to()` opens a builder with one setter per placeholder, and `publish()`
// compiles only once every one is bound - an unbound segment rides in the type, so the
// error names the segment that was forgotten
orders.message(&OrderPlaced { id }).to().tenant("acme").publish().await?;
// no declared name: `to(..)` takes a &str or a computed String
out.message(&Event { id }).to(format!("events.{shard}")).publish().await?;
// bytes as delivered: a Serialized newtype rides the same entry, and no codec runs
out.message(&Export(bytes)).to("egress").publish().await?;
```
A message declaring `headers = Meta` publishes only with `.with_headers(&meta)` - forgetting it,
or passing another type, does not compile. `with_headers` also takes an already-built `HeaderMap`
by value. A `Serialize` type owned by another crate cannot derive `Outgoing`: wrap it in a newtype
that does. The publisher may add a base of its own (`base_headers`, a tenant or a producer name);
the builder lays that base down first and writes the call site's headers over it, key by key.
## Publish transforms
A published message passes two levels. A `PublishTransform` sits on one position, chained with
`.transform(..)` after the `.out(..)` that named it; a `PublishLayer` sits on the application,
added with `.publish_layer(..)` before `with_broker`, and wraps every publish a handler makes,
replies and slots alike. A position's transforms run first, then the app-wide layers, then the
send. A router's slots run their own transforms and not the app-wide chain, because its routes are
typed before the app that mounts them.
Every impl of `PublishTransform<K, Options>` declares three things. What it **reads** is the
position's kind: `ForReply<C>` hands it `PublishContext<'_, C>` (the delivery being answered - its
channel, the incoming headers, the broker's typed per-delivery fields), `ForSlot` hands it
`SlotContext<'_>` (the slot's own name and nothing else, because the body issued that publish and
has already read its own `Context`). A transform reading nothing writes one impl over every kind
and mounts anywhere. What it **writes** is `Options`, the broker's per-message settings: a
transform generic over them touches none and mounts over any broker's publisher, one naming a
broker's type mounts over that broker's publisher and nowhere else. What it **may do** is
`type Destination`: `Reads` leaves the destination alone, `Names` sets it. Stable Rust has no
associated-type default, so every impl writes the line; almost every one writes `Reads`.
```rust
use ruststream::runtime::{
ContextKind, ForReply, ForSlot, Names, Outgoing, PublishContext, PublishTransform, Reads,
SlotContext,
};
/// Reads and writes nothing, so one impl serves a reply and a slot on any broker alike.
struct EnvelopeTransform;
impl<K: ContextKind, Options> PublishTransform<K, Options> for EnvelopeTransform {
type Destination = Reads;
fn apply(
&self,
out: &mut Outgoing<'_>,
_options: &mut Option<Options>,
_cx: &K::View<'_>,
) {
out.headers_mut().insert("x-envelope", b"1".to_vec());
}
}
/// Writes one of the broker's per-message settings, so it mounts over that broker's publisher
/// only. The position starts at `None` - the policy's defaults apply - and on a slot it starts
/// as a copy of what the call site's steps set, so this completes the call rather than
/// replacing it.
struct Expedite;
impl PublishTransform<ForSlot, MqttOptions> for Expedite {
type Destination = Reads;
fn apply(
&self,
_out: &mut Outgoing<'_>,
options: &mut Option<MqttOptions>,
_cx: &SlotContext<'_>,
) {
options.get_or_insert_with(MqttOptions::default).retain = Some(true);
}
}
```
### Naming a destination from a transform
Some answers have no destination to declare: an AMQP request carries the queue to answer on in its
`reply-to` header, a ZeroMQ `ROUTER` addresses each answer to the peer that asked. A transform
declaring `Destination = Names` sets the name per delivery, and it mounts with the ordinary
`.transform(..)` step - there is no second verb.
```rust
struct ReplyTo;
impl<C, Options> PublishTransform<ForReply<C>, Options> for ReplyTo {
type Destination = Names;
fn apply(
&self,
out: &mut Outgoing<'_>,
_options: &mut Option<Options>,
cx: &PublishContext<'_, C>,
) {
if let Some(to) = cx.headers().get_shared("reply-to")
&& let Ok(to) = Str::try_from(to)
{
out.set_name(to);
}
}
}
// `Response` leaves its destination open, so this position offers the naming right;
// "answers" is the declared fallback and what the document reports
b.include(answer).out_reply(Publish).transform(ReplyTo);
```
`get_shared` hands the header value over as the buffer the delivery holds, and `Str::try_from`
checks it is UTF-8, so the address reaches `set_name` without being copied.
A position offers that right only where nothing declared the destination, and elsewhere the
`.transform(..)` call does not compile, naming the type that already declared one. A reply type
carrying `#[outgoing(name = ..)]` offers nothing, a batch's replies never do (they answer many
deliveries at once), and a position offers it once. A slot offers it only when every type in its
marker's `#[publishes(..)]` dictionary leaves its destination open, which means a marker with no
dictionary - `DefaultSlot` included - never does. A slot that gave the right away offers plain
sending only: a transaction and a request / reply round trip do not compile on it.
## Broker settings per message
A QoS, a priority, an ordering key, an expiration are the broker's own per-message settings, not
headers. They reach a publish from two places. The publish policy carries the defaults, named at
the mount site, and every message through that position is sent with them:
```rust
b.include(record).out(Ledger, Publish::default().qos(1)).build();
b.include(acknowledge).out_reply(Publish::default().qos(1)); // a reply has no call site
```
The call adjusts one setting for one message, through a step the broker adds to the publish
builder, and a transform on the position adjusts one where there is no call site to (a reply, the
deferred retry copy). A setting neither names keeps the policy's default, and the rest of the
publish is still the mount site's: the entry's codec, its transforms, the slot the test harness
records it against.
```rust
// the body imports the broker's prelude for the step, and names its options type in the bound -
// the one place a handler body imports more than the framework prelude
use ruststream_rumqttc::prelude::*;
#[subscriber("telemetry.in")]
async fn record(
reading: &Reading,
Out(audit): Out<impl Publisher<Options = MqttOptions>, Telemetry>,
) -> HandlerOutcome {
if audit.message(reading).to("telemetry.out").retain(true).publish().await.is_err() {
return HandlerOutcome::retry();
}
HandlerOutcome::ack()
}
```
The steps are on the builder, so they work on every publish surface: an `Out` slot, a transaction
opened on one, a bare publisher from the state or a startup hook. Which steps exist is the broker's
documentation to answer. A setting never rides a header and never wraps the publisher (an adapter
would lose the mount site's codec).
## 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::memory::prelude::*;
#[derive(Clone)]
struct Database; // connection pool, etc.
#[derive(Clone, FromRef)]
struct AppState {
db: Database,
}
#[subscriber("orders")]
async fn handle(order: &Order, State(db): State<Database>) -> HandlerOutcome {
// db.store(order).await;
let _ = db;
HandlerOutcome::ack()
}
#[ruststream::app]
fn app() -> impl 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); })
}
```
## First publish at startup (`after_startup`)
A publish that has to happen once, at startup (seeding reference data, announcing the service),
goes through the scope's `after_startup`. It takes two arguments: a **publish policy** and a hook.
The runtime pairs the policy with the connected broker and hands the hook the **live** publisher,
so the seed cannot race the connect - there is no "not connected yet" case to handle. The hook
returns `Result<(), E>` for any `E: std::error::Error + Send + Sync`.
The policy is the broker's own - usually there is nothing to wrap:
```rust
use ruststream::memory::prelude::*;
// in with_broker(broker, |b| { ... }):
b.after_startup(Publish, async move |publisher| {
// `publisher` is the live MemoryPublisher, already connected
publisher
.message(&Region { code: "eu-west" })
.to("reference.regions")
.publish()
.await
});
```
Every `Publisher` gets `.message(&value)` from `PublishExt`, which the prelude carries - a
`Serialize` value encodes with `DefaultCodec`, a `Serialized` value's bytes leave as they are -
so a bare policy is all a hook ever takes. Name a codec at the call, and open a transaction
straight on the publisher:
```rust
use ruststream::codec::CborCodec;
// a codec other than the crate default, named at the publish
b.after_startup(Publish, async move |p| {
p.message(&Seed { id: 1 }).with_codec(CborCodec).publish().await
});
// an atomic seed: `begin()` opens the scope a transactional publisher owns
b.after_startup(
TransactionalPublish,
async move |seeder| seed_events(seeder).await.map_err(std::io::Error::other),
);
```
The hook's error type is a single concrete error, so a body that mixes error types collects them
in one function and the hook maps once, as above. The transaction scope owns the buffer, and both
records become visible together on commit or not at all - a commit without a begin, a second
commit, or a publish after settling do not compile:
```rust
use std::error::Error;
use ruststream::prelude::*;
// The bound names the capability the seeding needs, not the broker's publisher; the same
// capability unlocks `begin()` on an `Out` slot (`Out<impl TransactionalPublisher, Journal>`).
async fn seed_events<P>(seeder: P) -> Result<(), Box<dyn Error + Send + Sync>>
where
P: TransactionalPublisher,
{
let scope = seeder.begin().await?;
scope.message(&Event { id: 1 }).to("events").publish().await?;
scope.message(&Event { id: 2 }).to("events").publish().await?;
scope.commit().await?;
Ok(())
}
```
A policy is named the same way wherever a publisher is attached, and the verb is one:
`.out_reply(..)` for a reply handler, `.out(marker, ..)` for a slot, `.out_retry(..)` for the
deferred `retry_after` copy, `after_startup(policy, hook)` for the first publish.
## 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_rdkafka::prelude::*; // the context keys ride the broker's prelude too
#[subscriber("orders")]
async fn handle(order: &Order, Ctx(partition): Ctx<Partition>) -> HandlerOutcome {
tracing::info!(partition, "delivery");
HandlerOutcome::ack()
}
```
## Typed headers (`Headers<T>`)
`Headers<T>` parses the delivery headers into a flat contract struct before the body runs; a
missing or unparsable header settles the delivery by `on_failure(decode = ..)` (drop by default).
On a batch the batch pairs each element with its contract: take `&[Message<H, T>]` and read
`element.headers`. The untyped `name -> bytes` map itself is `HeaderMap` (`ctx.headers()`), and
`Headers` is only the extractor.
```rust
use ruststream::prelude::*;
#[subscriber("chunks")]
async fn convert(chunk: &Chunk, Headers(meta): Headers<ChunkMeta>) -> HandlerOutcome {
tracing::info!(meta.task_id, "converting");
HandlerOutcome::ack()
}
```
## Context (optional explicit parameter)
When you need the raw delivery objects, take `ctx: &mut Context<'_, C, S>` as the second parameter,
right after the payload (`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 (`HeaderMap`).
- `ctx.state()` - the shared application state built by `on_startup`.
- `ctx.context(KEY)` - one broker field by compile-time key, the `Ctx<K>` extractor's other half.
- `ctx.set(KEY, value)` - a per-delivery scratch value, for middleware.
- `ctx.after(outcome).then(fut)`, `ctx.after_ack(fut)`, `ctx.after_settle(fut)` - a post-settle
hook selected by how the delivery settles.
There is no `ctx.get::<T>()` and no named-publisher registry; state is typed (`State<T>` /
`ctx.state()`), and publishing goes through a `reply` clause or an `Out` parameter.
On a batch handler `C` is the broker's *subscription-scoped* context, not its per-delivery one:
name it (`ctx: &mut Context<'_, MemoryBatchContext>`) to read what the whole subscription shares -
a seek handle, say - which the runtime builds once per batch. A position or a header belongs to
one delivery, so it rides the batch's elements (`&[Message<H, T>]`) instead, and a batch body
naming the per-delivery context type does not compile.
## 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, JsonCodec};
use ruststream::prelude::*;
// every handler in this scope decodes with CborCodec
RustStream::new(info).with_broker_codec(broker, CborCodec, |b| {
b.include(handle);
});
// name the codec for one handler by mounting it through a router
b.include_router(Router::new().with_codec(JsonCodec).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
`Router<B, impl RouterDef<B>>`. Groups compose with `merge`. 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`; it wraps every publish a handler
makes, replies and `Out` slots alike, except in a router (its routes are typed before the app
that mounts them, so a router's slots carry their own `.transform(..)` steps only).
```rust
use ruststream::memory::prelude::*; // Router and RouterDef ride the prelude
fn orders() -> Router<MemoryBroker, impl RouterDef<MemoryBroker>> {
Router::new().include(accept)
}
fn shipping() -> Router<MemoryBroker, impl RouterDef<MemoryBroker>> {
Router::new().include(dispatch)
}
// in with_broker(broker, |b| { ... }):
b.include_router(orders().merge(shipping()));
```
On a router every registration that carries an attachment (a reply publisher, an `Out` slot) ends
with an explicit `.build()`: a chain without it never becomes a router, so it does not compile.
With no `.out_reply(..)` named, `.build()` takes the broker's own default publish policy.
## Broker-specific subscriptions
For options the by-name form can't express (NATS JetStream, Kafka consumer groups / start offsets),
name the broker's descriptor in the attribute - the broker's prelude already carries it. Broker
crates layer their own extension trait over the mount builder for the rest, so the settings read in
the broker's own vocabulary.
```rust
use ruststream_nats::prelude::*; // brings SubscribeOptions
#[subscriber(SubscribeOptions::new("orders.*").jetstream("ORDERS").durable("worker"))]
async fn handle(order: &Order) -> HandlerOutcome {
HandlerOutcome::ack()
}
```
## AsyncAPI metadata
Decoding only needs `serde::Deserialize`. The extra derives are **documentation only** for the
generated AsyncAPI spec and are optional: `MessageInfo` (in the prelude) 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::prelude::*;
use ruststream::schemars::JsonSchema;
use serde::Deserialize;
/// An order placed on the orders channel.
#[derive(Debug, Deserialize, MessageInfo, JsonSchema)]
struct Order {
id: u64,
}
```
## Testing
Use the `TestApp` harness (`testing` feature) on the service's production app, the builder `main`
runs: `TestApp::start` connects every broker in process, with no network and no server, injects
input through the broker, and exposes per-broker assertions addressed by the production broker
type. `TestApp::start_live` runs the same test body against a running stand. `MemoryBroker` is a
real broker here, driven through the same dispatch path as production.
```rust
use ruststream::memory::prelude::*;
use ruststream::testing::TestApp;
let tb = TestApp::start(service()).await?;
// the publish drives the handler to a standstill before returning; in a single-broker app
// `tb.message(..)` says the same without naming the broker
tb.broker::<MemoryBroker>()
.message(&Order { id: 42 })
.to("orders")
.publish()
.await?;
tb.broker::<MemoryBroker>()
.subscriber("orders")
.assert_called_once()
.with(&Order { id: 42 })
.settled(HandlerOutcome::ack());
tb.broker::<MemoryBroker>()
.published::<Receipt>("receipts")
.assert_called_once()
.with(&Receipt { order_id: 42 });
```
A handler that takes a value its broker's delivery already holds is asserted with
`with_value(&expected)` and `received_values::<T>()`: that value never was bytes, so `with` has
nothing to decode.
`tb.out::<Marker>()` is the other view: exactly what left through one injected publisher, across
every broker. It takes the same assertions as `published` (`assert_called_once`, `with_raw`,
`messages`; chain `.decoded_as::<T>()` for the typed `with`). Both views read back the per-message
broker settings the latest publish carried - `with_options(&expected)` compares them,
`assert_options_default()` states that no step touched one. They are recorded for what the runtime
publishes; a message the test published itself carries none, and asking for them panics.
```rust
// every field of an options type is optional, so the expectation is a struct literal
tb.out::<Telemetry>()
.assert_called_once()
.with_options(&MqttOptions { retain: Some(true), ..MqttOptions::default() });
tb.out::<Notes>().assert_called_once().assert_options_default();
```
A broker crate provides its in-process mode under its `testing` feature: enable it in
`[dev-dependencies]`, build the app on the production broker (`KafkaBroker::new(..)`), and address
it as `tb.broker::<KafkaBroker>()`. A test never names a stand-in type.