NatsBroker

NatsBroker connects a service to NATS; new only records the server addresses, and the runtime connects it at startup.

NatsBroker::new takes one address or a comma-separated list: nats://host:port, tls://host:port or a bare host:port. It does no I/O, so the whole service fits the synchronous #[ruststream::app] builder. Everything else the client needs, such as its name, its reconnects, credentials and TLS, is an async-nats setting attached with with_options.

At startup the runtime connects the broker and receives a ConnectedNatsBroker. The connect dials the server, so an unreachable server fails startup instead of the first publish. Every subscription and every publisher of the service opens against that connected form. At shutdown the runtime drains the connection and turns it into a ClosedNatsBroker.

#[subscriber("orders.created")] reads the Core NATS subject orders.created, outside any queue group. The descriptors for a queue group, a subject pattern or a JetStream stream are on the Subscriptions pages.

The crate has two features, both off by default. testing adds the in-process mode a TestApp test connects, and asyncapi adds the NATS bindings of the generated document.

Example

The service reads the subject orders.created. cargo run --example nats_core --features testing -- run starts it against a server on localhost:4222.

Macros
Manual

examples/nats_core.rs
use ruststream_nats::prelude::*;

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

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(handle);
        },
    )
}

examples/manual/nats_core.rs
use ruststream_nats::prelude::*;

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 received");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(subscriber("orders.created", Receive).build());
        },
    )
}

Example​

Example