Prelude

One glob, ruststream_nats::prelude, brings the core prelude, the broker, its descriptors and its publish policies under mount-site names.

use ruststream_nats::prelude::*; is the one import a NATS service writes. It carries the core's own prelude, so #[subscriber], RustStream, HandlerOutcome and Out come with it. It adds NatsBroker, the descriptors CoreSubject, CoreWildcard and JetStreamSubject, and NonZeroDuration for their time settings.

The publish policies come under the names a mount site uses on every broker. Publish is NatsPublish, plain Core NATS publishing. JetStreamPublish keeps its name, because JetStream is the concept and not this broker's word for one. Publish also makes requests, so the prelude carries the RequestReply capability and no separate Request policy. Two broker preludes in one file both bring a Publish, and such a file names the prefixed type from the crate root.

A handler body bounds its slots with capability traits, such as Out<impl Publisher> or Out<impl RequestReply>, and names no broker type. The one exception is a body that sets a per-message JetStream setting, such as a message id. It needs JetStreamPublishSteps in scope and the slot bounded Out<impl Publisher<Options = JetStreamOptions>, Marker>, both from this prelude. The keys of the JetStream context live in ruststream_nats::context::keys.

Example

The descriptors, their settings and the reply policy all come from the one glob.

Macros
Manual

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

#[subscriber(JetStreamSubject::new("orders.created", "ORDERS").durable("confirmer"), reply)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

#[subscriber(CoreSubject::new("orders.confirmed").queue_group("mailers"))]
async fn mail(confirmation: &Confirmation) -> HandlerOutcome {
    tracing::info!(confirmation.id, "confirmation mailed");
    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(confirm).out_reply(Publish);
            b.include(mail);
        },
    )
}

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

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

struct Mail;

impl Handle<Confirmation> for Mail {
    fn handle(
        &self,
        confirmation: &Confirmation,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(confirmation.id, "confirmation mailed");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            let stored = JetStreamSubject::new("orders.created", "ORDERS").durable("confirmer");
            b.include(subscriber(stored, Confirm).reply().build())
                .out_reply(Publish);
            let confirmed = CoreSubject::new("orders.confirmed").queue_group("mailers");
            b.include(subscriber(confirmed, Mail).build());
        },
    )
}

Example​

Example