Tutorial: build your first service¶
By the end of this page you have a running orders service: a message type, a handler, a reply, and a router that collects them. It runs on the in-memory broker, so there is nothing external to start. Swapping in a real broker is a one-line change, and step 7 shows it.
1. Create the crate¶
[package]
name = "orders-service"
version = "0.1.0"
edition = "2024"
[dependencies]
ruststream = { version = "0.7", 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 subscriber definition named after the function.
use ruststream::runtime::HandlerOutcome;
use ruststream::schemars::JsonSchema;
use ruststream::{Outgoing, subscriber};
use serde::{Deserialize, Serialize};
/// An order placed by a customer.
#[derive(Debug, Deserialize, JsonSchema)]
pub(crate) struct Order {
pub(crate) id: u64,
pub(crate) quantity: u32,
}
#[subscriber("orders")]
pub(crate) async fn handle(order: &Order) -> HandlerOutcome {
println!("order {} x{}", order.id, order.quantity);
HandlerOutcome::ack()
}
use std::future::{Future, ready};
use ruststream::prelude::*;
use ruststream::schemars::JsonSchema;
use serde::Deserialize;
/// An order placed by a customer.
#[derive(Debug, Deserialize, JsonSchema)]
pub(crate) struct Order {
pub(crate) id: u64,
pub(crate) quantity: u32,
}
/// The handler: `#[subscriber("orders")]` generates this struct and this impl. The subject it
/// carried is named where the handler is mounted instead.
pub(crate) struct Receive;
impl Handle<Order> for Receive {
fn handle(
&self,
order: &Order,
_outs: &(),
_ctx: &mut Context<'_>,
) -> impl Future<Output = Result<(), HandlerOutcome>> {
println!("order {} x{}", order.id, order.quantity);
ready(Ok(()))
}
}
A handler returns a HandlerOutcome: an ack, or a nack that
drops or requeues the message. You can return () or Result<(), E> instead, where Ok acks and
Err drops.
The JsonSchema derive puts the payload's schema into the AsyncAPI document of step 6. The type's
doc comment becomes the message description there. You need no extra dependency for it: the
asyncapi feature re-exports schemars.
3. Wire it into an app¶
mod orders;
use std::error::Error;
use ruststream::memory::prelude::*;
use crate::orders::Receive;
fn app() -> RustStream {
RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
b.include(subscriber("orders", Receive).build());
})
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
app().run().await?;
Ok(())
}
Codec defaults
include decodes with the default codec, so it needs no codec argument. The default is json
when the feature is enabled, otherwise cbor, otherwise msgpack. You can set one codec for
all the broker's handlers at once with with_broker_codec(broker, codec, |b| ...). See
Codecs for the full resolution rules.
Run it:
4. Reply to messages¶
To publish a reply, return the reply value and write publish on the subscriber. The Outgoing
derive on the reply type says where it goes:
/// The service's answer to an order.
#[derive(Debug, Serialize, JsonSchema, Outgoing)]
#[outgoing(name = "confirmations")]
pub(crate) struct Confirmation {
pub(crate) id: u64,
pub(crate) accepted: bool,
}
#[subscriber("orders", publish)]
pub(crate) async fn confirm(order: &Order) -> Confirmation {
Confirmation {
id: order.id,
accepted: order.quantity > 0,
}
}
use serde::Serialize;
/// The service's answer to an order, published on `confirmations`.
#[derive(Debug, Serialize, JsonSchema)]
pub(crate) struct Confirmation {
pub(crate) id: u64,
pub(crate) accepted: bool,
}
/// What `#[derive(Outgoing)]` with `#[outgoing(name = "confirmations")]` writes: the destination
/// is fixed on the type, so nothing else names it.
impl OutgoingDestination for Confirmation {
type Form = FixedName;
const DESTINATION: &'static str = "confirmations";
}
impl MessageHeaders for Confirmation {
type Contract = NoHeaders;
}
/// The reply form of the same trait: the second parameter of `Handle` is the reply type, so the
/// body returns a `Confirmation`, and the mount site supplies the publisher it leaves through.
pub(crate) struct Confirm;
impl Handle<Order, Confirmation> for Confirm {
fn handle(
&self,
order: &Order,
_outs: &(),
_ctx: &mut Context<'_>,
) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
ready(Ok(Confirmation {
id: order.id,
accepted: order.quantity > 0,
}))
}
}
Mount confirm next to handle with the same include. The reply is published with the broker's
default publish policy and encoded with the default codec.
use crate::orders::{Confirm, Receive};
fn app() -> RustStream {
RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
b.include(subscriber("orders", Receive).build());
// The definition names the reply's destination; with no `.out_reply(..)` at the mount
// site the reply leaves through the broker's default publisher.
b.include(
subscriber("orders", Confirm)
.reply()
.to("confirmations")
.build(),
);
})
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
app().run().await?;
Ok(())
}
Publishing from inside a handler and the other ways to publish are in Publishing & replies.
5. Organize with a router¶
As the number of handlers grows, keep them in their own module and collect them into a
Router:
use ruststream::memory::prelude::*;
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> {
Router::new()
.include(orders::handle)
.include(orders::confirm)
.out_reply(Publish)
.build()
}
use ruststream::memory::prelude::*;
use crate::orders::{Confirm, Receive};
// Each handler is bound to its subject where it is mounted; the definition says what it replies
// with, the reply type says where it goes, and the mount chain names who publishes it. The
// publisher wiring is still a publish policy - pure declaration, so the router needs no broker
// at all.
pub(crate) fn orders() -> impl RouterDef<MemoryBroker> {
Router::new()
.include(subscriber("orders", Receive).build())
.include(subscriber("orders", Confirm).reply().build())
.out_reply(Publish)
.build()
}
include adds a plain handler to the router directly. A handler that publishes a reply hands back
a mount chain instead: .out_reply(..) names the reply's publish policy, and .build() finishes
the registration. Without .out_reply(..), .build() takes the broker's default publish policy -
the same one include took in step 4. Routing covers the rest of the router
surface.
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);
})
}
mod orders;
mod routes;
use std::error::Error;
use ruststream::memory::prelude::*;
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);
})
}
// What `#[ruststream::app]` wraps around the builder: the runtime entry point. Its CLI
// (`run`, `asyncapi gen`) is what a hand-written `main` gives up.
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
app().run().await?;
Ok(())
}
6. Inspect the AsyncAPI document¶
Every subscriber adds a channel and a receive operation to the document. handle and confirm
share the orders channel and get one operation each, because their subscriptions are separate. The
reply adds a send operation on confirmations.
The document keeps the payload schemas under components.messages. 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 the swap is
a one-line change. Add the broker crate as a dependency and construct it there instead of
MemoryBroker::new(), for example NatsBroker::new("nats://localhost:4222"). The handlers, the
router and the codecs stay as they are. Brokers lists the
available brokers and the swap for each of them.
The complete service is a compiled example
Every snippet on this page comes from
examples/tutorial
in the repository, which CI builds on every change. first_app.rs and reply_app.rs are the
service as steps 3 and 4 leave it, and main.rs is the finished one. Run it yourself with
cargo run --example tutorial --features macros,memory,json,asyncapi -- 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.