Struct JetStreamSubject 

pub struct JetStreamSubject { /* private fields */ }

A subject read through a JetStream pull consumer on one stream.

The consumer is created when the subscription opens. Name it with durable and the server keeps its position across restarts; leave the name out and the consumer is ephemeral, discarded when the subscriber drops.

Examples

use ruststream_nats::prelude::*;
use serde::Deserialize;

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

// The durable keeps its position across restarts, and every replica shares it.
#[subscriber(
    JetStreamSubject::new("orders.*", "ORDERS")
        .durable("billing")
        .ack_wait(NonZeroDuration::from_secs(nonzero!(30)))
)]
async fn bill(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "billed");
    HandlerOutcome::ack()
}

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

A queue group is Core NATS only, so it is not a method here:

use ruststream_nats::JetStreamSubject;

let bad = JetStreamSubject::new("orders.*", "ORDERS").queue_group("workers");

Implementations

impl JetStreamSubject

pub fn new(subject: impl Into<String>, stream: impl Into<String>) -> Self

Reads subject from the stream named stream.

subject is the pattern the subscription reports and the default consumer filter; stream is the JetStream stream that stores it.

pub fn durable(self, name: impl Into<String>) -> Self

Names the consumer, so the server keeps its position across restarts.

pub fn filter_subject(self, subject: impl Into<String>) -> Self

Narrows which subjects of the stream this consumer reads. Defaults to the subject the constructor named.

pub const fn ack_wait(self, ack_wait: NonZeroDuration) -> Self

Per-message acknowledgement timeout. After this window an unacked delivery is redelivered. Defaults to 30 seconds.

pub const fn max_ack_pending(self, max: i64) -> Self

Soft cap on in-flight unacked deliveries. Defaults to 1024.

pub const fn deliver_policy(self, policy: DeliverPolicy) -> Self

Where a newly created consumer starts reading. Defaults to DeliverPolicy::All.

pub const fn pull_expires(self, expires: NonZeroDuration) -> Self

How long one fetch waits before delivering a partial (or retrying an empty) batch. Defaults to 5 seconds. Has no effect on the per-message Subscriber::stream path.

How many messages a batch carries is not set here: the batch size travels with the registration (b.include(handler.batch(nonzero!(6)))) and reaches the fetch as the argument of BatchSubscriber::batches.

Trait Implementations

impl Clone for JetStreamSubject

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for JetStreamSubject

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl IntoSource for JetStreamSubject

type Source = JetStreamSubject

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl NatsSubscription for JetStreamSubject

fn subject(&self) -> &str

The subject pattern this subscription receives messages on.

fn ensure_subject(&self) -> Result<(), NatsError>

The one thing about a subscription the types cannot settle: the subject is a string the caller supplies, often from configuration, and a descriptor constructor is the expression a #[subscriber(..)] attribute writes, so it cannot hand back a Result. Every subscribe implementation calls this before it does any work.

impl RedeliveryAddressed<ConnectedNatsBroker> for JetStreamSubject

The consumer’s filter subject, which defaults to the subject the descriptor names.

A JetStream consumer holds a delivery itself and takes the delay in its negative acknowledgement, so the framework publishes no copy here and this address is what the conformance suite checks rather than what a running service uses. A consumer filtering a pattern reports that pattern, which is not a publish destination; nothing publishes to it, and a .to(name) at the mount site overrides it.

fn redelivery_address( &self, _connected: &ConnectedNatsBroker, ) -> impl Future<Output = Result<RedeliveryAddress, NatsError>> + Send

Where a publish reaches this subscription again.

impl SubscriptionSource<ConnectedNatsBroker> for JetStreamSubject

type Subscriber = NatsSubscriber

The subscriber type this source opens.

type Copies = AddressedCopies

Who publishes the copies this subscription’s retries are made of, and who names where they go: AddressedCopies, NamedCopies or BrokerMoves.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe( self, connected: &ConnectedNatsBroker, ) -> Result<Self::Subscriber, NatsError>

Opens the subscription against the connected broker. Called once at startup.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

where Self: Sized,

Takes the registration’s retry declaration into the descriptor, before the subscription opens.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.