JetStreamPublish

JetStreamPublish stores each message in a stream and waits for the stream's acknowledgement, so a refused publish is an error.

A mount site names it like the plain policy: .out_reply(JetStreamPublish::default()), .out(Slot, JetStreamPublish::default()), or b.after_startup(..). The runtime pairs it into a JetStreamPublisher at startup. Every publish waits for the stream that stores the subject to acknowledge the message. A subject no stream stores, a violated expectation or a storage failure fails the publish with NatsError::JetStream. So the publish costs a round trip to the server, and the handler learns whether the message was stored. A plain Publish to a subject a stream stores is stored too, but the publisher does not learn whether it was. The policy can name the stream it expects with expect_stream. One message states its own settings with publish steps: a deduplication id with message_id and the position it expects the stream at with the expectations. publish_ack returns the acknowledgement itself. A reply can be stored in a stream while the rest of the service publishes over Core NATS: each mount site chooses its policy.

Example

The confirmation an order is answered with is stored in a stream of its own.

Macros

examples/nats_jetstream.rs

/// The confirmation an order is answered with. Its subject is outside the `ORDERS` stream, so a
/// confirmation is never read back as an order.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "confirmations")]
struct Confirmation {
    id: u64,
}

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

Manual

examples/manual/nats_jetstream.rs

/// The confirmation an order is answered with. Its subject is outside the `ORDERS` stream, so a
/// confirmation is never read back as an order.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "confirmations")]
struct Confirmation {
    id: u64,
}

struct Confirm;

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

The mount site names the JetStream policy for the reply, for the archive slot and for the startup publish.

Macros

examples/nats_jetstream.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            // The reply is stored in a stream and acknowledged by it; the rest of the service
            // can keep publishing over Core NATS.
            b.include(confirm)
                .out_reply(JetStreamPublish::default().expect_stream("CONFIRMATIONS"));
            b.include(archive)
                .out(
                    Archive,
                    JetStreamPublish::default().expect_stream("ARCHIVE"),
                )
                .build();
            b.after_startup(JetStreamPublish::default().expect_stream("ORDERS"), seed);
        },
    )
}

Manual

examples/manual/nats_jetstream.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            let created = JetStreamSubject::new("orders.created", "ORDERS").durable("confirmer");
            let shipped = JetStreamSubject::new("orders.shipped", "ORDERS").durable("archiver");
            // The reply is stored in a stream and acknowledged by it; the rest of the service
            // can keep publishing over Core NATS.
            b.include(subscriber(created, Confirm).reply().build())
                .out_reply(JetStreamPublish::default().expect_stream("CONFIRMATIONS"));
            b.include(subscriber(shipped, ArchiveOrder).build())
                .out(
                    Archive,
                    JetStreamPublish::default().expect_stream("ARCHIVE"),
                )
                .build();
            b.after_startup(JetStreamPublish::default().expect_stream("ORDERS"), seed);
        },
    )
}