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

cargo new orders-service
cd orders-service

Cargo.toml

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

[dependencies]
ruststream = { version = "0.7", features = ["macros", "memory", "json", "asyncapi", "logging"] }
serde = { version = "1", features = ["derive"] }
tracing = "0.1"

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.

Macros

src/orders.rs

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 {
    tracing::info!(order.id, order.quantity, "order received");
    HandlerOutcome::ack()
}

Manual

src/orders.rs

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>> {
        tracing::info!(order.id, order.quantity, "order received");
        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

Macros

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-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(handle);
    })
}

Manual

src/main.rs

mod orders;

use std::process::ExitCode;

use ruststream::memory::prelude::*;
use ruststream::runtime::cli::run_main;

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());
    })
}

fn main() -> ExitCode {
    run_main(app)
}

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:

cargo run -- run

The logging feature prints the service's log to the terminal: the startup lines, then a line for each order the handler logs with tracing::info!.

4. Reply to messages

To publish a reply, return the reply value and write reply on the subscriber. The Outgoing derive on the reply type says where it goes:

Macros

src/orders.rs

/// 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", reply)]
pub(crate) async fn confirm(order: &Order) -> Confirmation {
    Confirmation {
        id: order.id,
        accepted: order.quantity > 0,
    }
}

Manual

src/orders.rs

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.

Macros

src/main.rs

use crate::orders::{confirm, handle};

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

Manual

src/main.rs

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(),
        );
    })
}

fn main() -> ExitCode {
    run_main(app)
}

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:

Macros

src/routes.rs

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()
}

Manual

src/routes.rs

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.

Macros

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);
    })
}

Manual

src/main.rs

mod orders;
mod routes;

use std::process::ExitCode;

use ruststream::memory::prelude::*;
use ruststream::runtime::cli::run_main;

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]` writes around the builder: the `run` and `asyncapi gen` commands.
fn main() -> ExitCode {
    run_main(app)
}

6. Inspect the AsyncAPI document

cargo run -- asyncapi gen

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,logging -- 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.