expect_stream

expect_stream names the stream a publisher's subjects must be stored by; a subject another stream stores, or none, is refused.

It is part of the policy, so it applies to every message the publisher sends. Each message carries the stream's name in the Nats-Expected-Stream header, and the server compares it with the stream that stores the subject. A mismatch fails the publish with NatsError::JetStream, and nothing is stored. A subject computed at run time, from a region, a tenant or a configuration value, is where this catches a mistake: a message routed to the wrong stream, or to none, is an error instead of a write somewhere else. With the asyncapi feature the generated document reports the expected stream on the channel, under x-ruststream-jetstream.

Example

Every shipment must be stored by SHIPMENTS; a region whose subject that stream does not store is refused, and the order is dropped.

Macros
Manual

examples/nats_publishing_expect_stream.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            // Every shipment must be stored by SHIPMENTS; a subject it does not store is an
            // error of the publish, not a message written somewhere else.
            b.include(ship)
                .out(
                    Shipments,
                    JetStreamPublish::default().expect_stream("SHIPMENTS"),
                )
                .build();
            b.include(dispatch);
        },
    )
}

examples/manual/nats_publishing_expect_stream.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            let paid = JetStreamSubject::new("orders.paid", "ORDERS").durable("shipper");
            let shipments = JetStreamSubject::new("shipments.*", "SHIPMENTS").durable("dispatcher");
            // Every shipment must be stored by SHIPMENTS; a subject it does not store is an
            // error of the publish, not a message written somewhere else.
            b.include(subscriber(paid, Ship).build())
                .out(
                    Shipments,
                    JetStreamPublish::default().expect_stream("SHIPMENTS"),
                )
                .build();
            b.include(subscriber(shipments, Dispatch).build());
        },
    )
}

Example​

Example