The service and its first subscriber

The shop starts as one subscriber that takes every placed order, mounted on the in-memory broker by a small app.

A RustStream service is an application: a builder that names its brokers and mounts subscribers on them. A subscriber is an async fn whose first parameter is the message, and #[subscriber] names the topic it reads. The runtime decodes the payload before the body runs, with JSON by default. What the body returns settles the delivery: HandlerOutcome::ack tells the broker the message is done. The in-memory broker carries messages inside the process, so the shop needs nothing installed. Chapter 11 swaps it for a networked broker, and the subscribers stay as they are.

The order

An order is a serde type. The Outgoing derive says where an order is published: orders.placed. The shop does not publish orders yet; a test does, and in chapter 10 the HTTP storefront does.

Macros

examples/guide/first.rs

/// An order placed in the storefront.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders.placed")]
struct Order {
    id: u64,
    customer: String,
}

Manual

examples/manual/guide/first.rs

/// An order placed in the storefront.
#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Order {
    id: u64,
    customer: String,
}

// What `#[derive(Outgoing)]` with `#[outgoing(name = "orders.placed")]` writes.
impl OutgoingDestination for Order {
    type Form = FixedName;

    const DESTINATION: &'static str = "orders.placed";
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

The manual form writes the two impls the derive generates: the destination, and the headers an order carries (none yet).

The first subscriber

accept reads orders.placed and logs each order through tracing. The manual form is a named type with an impl Handle<Order>: Ok(()) acks, and Err(outcome) settles by that outcome.

Macros

examples/guide/first.rs

/// Takes every placed order; for now it only logs it.
#[subscriber("orders.placed")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order = order.id, customer = %order.customer, "order accepted");
    HandlerOutcome::ack()
}

Manual

examples/manual/guide/first.rs

/// Takes every placed order; for now it only logs it.
struct Accept;

impl Handle<Order> for Accept {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order = order.id, customer = %order.customer, "order accepted");
        ready(Ok(()))
    }
}

The app

RustStream is the app; with_broker registers a broker and mounts subscribers on it with include. The builder only records what it is given: the broker connects when the service starts. The #[ruststream::app] attribute turns the builder into main, with the run and asyncapi gen commands. The manual form writes that main by hand: run_main over the builder.

Macros

examples/guide/first.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(accept);
    })
}

Manual

examples/manual/guide/first.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("orders.placed", Accept).build());
    })
}

/// What `#[ruststream::app]` writes: the `run` and `asyncapi gen` commands over `app`.
fn main() -> ExitCode {
    run_main(app)
}

Run it

cargo run --example guide_first --features testing,macros,memory,json,logging -- run

The logging feature prints the service's log to the terminal. The service connects the broker, opens the subscription and runs until Ctrl+C:

INFO ruststream::lifecycle starting service service=shop version=0.1.0 brokers=1 subscribers=1
INFO ruststream::lifecycle broker connected broker="ruststream::memory::MemoryBroker"
INFO ruststream::dispatch subscriber started subscriber=orders.placed input="guide_first::Order"
INFO ruststream::lifecycle service running subscribers=1

The rest of the guide grows this file into the shop of examples/guide.