Channels

A JetStream consumer, and a publisher that requires a stream, describe themselves under x-ruststream-jetstream on the channel.

A channel's address is the subject. A channel read by a JetStreamSubject reports the consumer's settings:

Field	Value
stream	the stream the consumer reads
durable	the durable name, when the descriptor sets one
filterSubject	the subjects of the stream the consumer reads, the descriptor's subject by default
ackWaitSeconds	how long the server waits for a settlement
maxAckPending	how many unacknowledged deliveries the consumer may hold
deliverPolicy	where a new consumer starts: all, new, last, last_per_subject, by_start_sequence or by_start_time

A channel a JetStreamPublish with expect_stream publishes to reports expectedStream and subject. The subject is the destination the mount site resolved, the one the stream has to store. A Core NATS subscription, a JetStream publish that requires no stream and a plain Publish add nothing to the channel.

Example

The channel orders.stored reports the durable consumer archiver of the stream ORDERS. The channel receipts.issued reports that the stream RECEIPTS has to store what is published there.

Macros

examples/nats_asyncapi.rs

#[subscriber(
    JetStreamSubject::new("orders.stored", "ORDERS")
        .durable("archiver")
        .ack_wait(NonZeroDuration::from_secs(nonzero!(10)))
        .max_ack_pending(64)
)]
async fn archive(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order archived");
    HandlerOutcome::ack()
}

#[subscriber("orders.paid", reply("receipts.issued"))]
async fn issue(order: &Order) -> Receipt {
    Receipt { id: order.id }
}

Manual

examples/manual/nats_asyncapi.rs

/// The durable consumer `archiver` on the stream `ORDERS`.
fn stored() -> JetStreamSubject {
    JetStreamSubject::new("orders.stored", "ORDERS")
        .durable("archiver")
        .ack_wait(NonZeroDuration::from_secs(nonzero!(10)))
        .max_ack_pending(64)
}

/// Receipts are stored by the stream `RECEIPTS`, or the publish fails.
fn receipts() -> JetStreamPublish {
    JetStreamPublish::default().expect_stream("RECEIPTS")
}