Application

A RustStream value names the service, holds its brokers and mounts each handler on the broker it subscribes to.

AppInfo holds the service's title and version, and AppInfo::new takes them. They name the service in its logs and in its AsyncAPI document. with_broker registers one broker and mounts handlers on it inside the closure, a BrokerScope. A service with several brokers calls it once per broker. A broker constructor does no I/O: the application connects every broker when it starts. Handlers mounted with b.include(..) decode with the default codec; with_broker_codec names another one for the whole scope. register_broker adds a broker that only publishes and has no handlers. A builder function returns impl App, so its signature stays the same as layers and state are added.

Example

Macros

examples/quickstart.rs

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

#[subscriber("orders")]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "got order");
    HandlerOutcome::ack()
}

examples/quickstart.rs

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

Manual

examples/manual/quickstart.rs

#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct Order {
    id: u64,
}

/// The handler: `#[subscriber("orders")]` generates this type and this impl. Every axis of the
/// form - the reply, the injections, the broker context, the application state - is a defaulted
/// parameter of `Handle`, so a plain body names none of them.
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, "got order");
        ready(Ok(()))
    }
}

examples/manual/quickstart.rs

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

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