Struct NatsSubscriber 

pub struct NatsSubscriber { /* private fields */ }

A NATS subscription.

Backed transparently by either a Core subscription (no ack) or a JetStream pull consumer (full ack/nack/term). Construct via ConnectedNatsBroker::subscribe_with with CoreSubject or JetStreamSubject, or let the runtime resolve either source at startup.

A handler over &[T] takes its deliveries in batches, and the mount site sets the largest batch with batch(n). On JetStream one batch is one pull request for at most n messages, and pull_expires closes a short one early. Core NATS has no batch on the wire, so the client assembles one and closes a partial batch 10 ms after its first delivery.

Examples

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

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

#[subscriber(JetStreamSubject::new("orders.*", "ORDERS").durable("orders-worker"))]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "handled");
    HandlerOutcome::ack()
}

#[subscriber(JetStreamSubject::new("orders.*", "ORDERS").durable("orders-reconciler"))]
async fn reconcile(orders: &[Order]) -> HandlerOutcome {
    tracing::info!(count = orders.len(), "reconciled");
    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);
            // One pull request fetches at most six messages.
            b.include(reconcile.batch(nonzero!(6)));
        },
    )
}

Trait Implementations

impl BatchSubscriber for NatsSubscriber

fn batches( &mut self, size: NonZeroUsize, ) -> impl Stream<Item = Result<Self::Batch, Self::Error>> + Send + '_

Returns a stream of batches of at most size messages.

size is the registration’s batch size, and each transport spends it in its own currency. JetStream batches on the wire: one stream item is one pull fetch of up to size messages, waiting at most pull_expires before delivering a partial batch (an empty fetch is retried, so the stream never yields empty batches). Core NATS has no wire-level batching, so its batches are assembled on the client by the framework’s BufferedSubscriber: a batch closes at size deliveries or 10 ms after its first one, whichever comes first.

Drive a subscriber through either Subscriber::stream or batches, not both at once: on JetStream each issues its own pull requests, so deliveries would be split between them.

Cancel safety

Dropping the returned stream between items is allowed. On JetStream, dropping it mid-fetch can leave already-fetched, undelivered messages to be redelivered after the consumer’s ack_wait.

type Batch = Vec<NatsMessage>

Container yielded by batches. Implementations choose between Vec, custom iterators, or anything else that yields the underlying Subscriber::Message.

impl Debug for NatsSubscriber

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

Formats the value using the given formatter.

impl Subscriber for NatsSubscriber

type Message = NatsMessage

The message type yielded by this subscriber.

type Error = NatsError

The error type yielded by the stream when delivery fails.

fn stream( &mut self, ) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_

Returns a stream of broker deliveries.