Tutorial: build your first service

This tutorial builds an orders service from scratch, explaining each piece. It uses the in-memory broker so there is nothing external to run; swapping in a real broker is a one-line change covered at the end.

1. Create the crate

cargo new orders-service
cd orders-service

Cargo.toml

[package]
name = "orders-service"
version = "0.1.0"
edition = "2024"

[dependencies]
ruststream = { version = "0.6", features = ["macros", "memory", "json", "asyncapi"] }
serde = { version = "1", features = ["derive"] }

2. Define a message and a handler

A handler is an async fn whose first parameter is the decoded payload. The #[subscriber] macro turns it into a mountable definition named after the function.

src/orders.rs

use ruststream::runtime::HandlerResult;
use ruststream::subscriber;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize)]
pub(crate) struct Order {
    pub(crate) id: u64,
    pub(crate) quantity: u32,
}

#[subscriber("orders")]
pub(crate) async fn handle(order: &Order) -> HandlerResult {
    println!("order {} x{}", order.id, order.quantity);
    HandlerResult::Ack
}

A handler returns a HandlerResult: Ack, or a nack that drops or requeues the message. Returning () or Result<(), E> also works - they convert into a result (Ok acks, Err drops).

3. Wire it into an app

src/main.rs

mod orders;

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, RustStream};

use crate::orders::handle;

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(MemoryBroker::new(), |b| b.include(handle))
}

The macro turns handle into a value named after the function, so you import and pass it directly.

Codec defaults

include decodes with the default codec - json if enabled, otherwise cbor, otherwise msgpack - so it needs no codec argument. To decode with a different one everywhere, set it once with with_broker_codec(broker, codec, |b| ...). See Codecs for the full resolution rules.

Run it:

cargo run -- run

4. Reply to messages

To publish a reply, return the reply value and name the destination with publish(..):

src/orders.rs

#[derive(Debug, Serialize)]
pub(crate) struct Confirmation {
    pub(crate) id: u64,
    pub(crate) accepted: bool,
}

#[subscriber("orders", publish("confirmations"))]
pub(crate) async fn confirm(order: &Order) -> Confirmation {
    Confirmation {
        id: order.id,
        accepted: order.quantity > 0,
    }
}

Mount it with plain include; the reply goes out through the broker's default publish policy under the default codec (chain .publisher(..) with a TypedPublisher stack to name a reply codec or add transforms):

// inside with_broker(...), with `confirm` imported from the orders module
b.include(confirm);

See Publishing & replies for the full picture, including publishing from inside a handler.

5. Organize with a router

As handlers grow, keep them in their own module and collect them into a Router:

src/routes.rs

use ruststream::memory::{MemoryBroker, MemoryPublish};
use ruststream::runtime::{Router, RouterDef, TypedPublisher};

use crate::orders;

// The reply wiring is a publish policy: pure declaration, so the router needs no broker at all.
pub(crate) fn orders() -> impl RouterDef<MemoryBroker> {
    let replies = TypedPublisher::new(MemoryPublish);
    Router::new()
        .include_publishing(orders::confirm, replies)
        .include(orders::handle)
}

src/main.rs

mod orders;
mod routes;

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, RustStream};

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        let router = routes::orders();
        b.include_router(router);
    })
}

6. Inspect the AsyncAPI document

cargo run -- asyncapi gen

Every subscriber becomes a channel and a receive operation; payload types that derive schemars::JsonSchema also contribute schemas. The output flags (-o, --yaml) and the document itself are covered in AsyncAPI.

7. Swap in a real broker

Nothing above is tied to the in-memory broker. The broker is chosen at with_broker, so swapping is a one-line change: add the broker crate as a dependency and construct it there (for example NatsBroker::new("nats://localhost:4222") instead of MemoryBroker::new()); the handlers, router, and codecs are unchanged. The available brokers and the side-by-side swap for each of them are in Brokers.

The complete service is a compiled example

Every snippet on this page is embedded from examples/tutorial in the repository, which CI builds on every change. Run it yourself with cargo run --example tutorial --features macros,memory,json -- run.

Next steps

• Middleware - cross-cutting logic around handlers.
• Lifespan - shared state and startup/shutdown hooks.
• Testing - test the handlers you just wrote, in-process.
• Metrics - Prometheus counters and histograms.