Crate ruststream_nats

Expand description

NATS broker for the RustStream messaging framework, backed by async-nats: Core NATS subjects and JetStream streams over one connection.

Core NATS hands a message to whoever is subscribed at that moment and stores nothing. JetStream stores it in a stream, which a pull consumer reads at its own pace and acknowledges message by message. Both models run on the same connection, so a service chooses per handler and per publish.

§A service

A handler is an async fn over the decoded payload, mounted on a broker inside the application object, and the attribute writes the main:

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

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

#[subscriber("orders.created")]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    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);
        },
    )
}

cargo run -- run starts it. NatsBroker::new only records the configuration; the runtime connects the broker at startup and drains it when the service stops.

§Where things are

The guides take a NATS service from installation to production, in this order: getting started, subjects, wildcards and queue groups, streams and consumers, acknowledgement and delivery guarantees, publishing to streams, retries and dead letters, request-reply, testing and connecting to production. The architecture pages show how the crate sits between async-nats and the runtime, and the benchmarks what it costs.

Each topic below links to its component page on the site, which shows it in a service; the item’s own documentation keeps the detail.

§Broker

NatsBroker: NatsBroker, connected into ConnectedNatsBroker and closed into ClosedNatsBroker.
ConnectOptions: NatsBroker::with_options attaches the async-nats connection options.
Security: credentials and TLS as connection options.
Startup: the connect within the client’s connection_timeout, and each subscription opened against the server.
Shutdown: the drain, and NatsError::Closed for a publisher used afterwards.
Prelude: prelude, the one glob a NATS service imports.
Errors: NatsError.

§Subscriptions

Subscriptions: a bare subject in #[subscriber] reads that Core NATS subject.
CoreSubject: CoreSubject, one Core NATS subject.
CoreWildcard: CoreWildcard, every subject a * or > pattern matches.
queue_group: CoreSubject::queue_group, one member of a group takes each message.
JetStreamSubject: JetStreamSubject, a stream read through a pull consumer.
durable: JetStreamSubject::durable, a named consumer that keeps its position.
filter_subject: JetStreamSubject::filter_subject, the subjects of the stream a consumer reads.
DeliverPolicy: DeliverPolicy, where a new consumer starts.
max_ack_pending: JetStreamSubject::max_ack_pending, the cap on unacknowledged deliveries.
Batches: a slice handler over NatsSubscriber batches, sized at the mount site and timed by JetStreamSubject::pull_expires.
Keys: PARTITION_KEY_HEADER keys the lanes of workers(n, by_key).

§Settlement

Acknowledgement: what an ack, a retry and a drop do on a JetStream delivery.
ack_wait: JetStreamSubject::ack_wait, how long the server waits for a settlement.
At most once: a Core NATS delivery, which nothing acknowledges.
Retries: max_attempts, counted by the server on JetStream and in a header on Core NATS.
retry_after: a delay the server holds on JetStream, and a copy the framework publishes on Core NATS.
Dead letters: dead_letter, the subject a spent delivery goes to.
out_retry: the subject a CoreWildcard registration sends its delayed copies to.

§Context

JetStreamContext: JetStreamContext and the keys in context::keys: stream, consumer, sequences, delivery count and pending count of a delivery.

§Publishing

NatsPublish: NatsPublish, the plain Core NATS policy, and its live NatsPublisher.
Headers: what a header must be to travel over NATS.
JetStreamPublish: JetStreamPublish and its live JetStreamPublisher, which wait for the stream’s acknowledgement.
expect_stream: JetStreamPublish::expect_stream, the stream a publisher’s subjects must be stored by.
message_id: JetStreamPublishSteps::message_id, a deduplication id carried in JetStreamOptions.
Expectations: the expect_last_* steps of JetStreamPublishSteps.
publish_ack: JetStreamPublisher::publish_ack and its PublishAck.

§Request-reply

RequestReply: request on NatsPublisher, answered on an inbox.
Responders: a handler that answers on the inbox of each request.

§Tooling

NATS bindings: the server, channels, operations and replies of the generated AsyncAPI document (feature asyncapi).
Testing: the in-process NATS server TestApp connects the broker to (feature testing), and Live runs against a running server.

§Cargo features

Both are off by default and additive. testing adds the in-process mode TestApp::start connects the broker through, and asyncapi contributes the NATS bindings to the generated document. The codecs, the macros and the harness itself are features of the core crate.

Modules§

context: Native JetStream metadata of a delivery, read by a handler through Ctx<K>.
prelude: The imports a service on NATS writes every time, in one glob.

Structs§

ClosedNatsBroker: The terminal witness returned by shutting down a ConnectedNatsBroker.
ConnectedNatsBroker: The typed witness that Broker::connect succeeded: holds the live connection.
CoreMessage: Wrapper around an async_nats::Message from a core (non-JetStream) subscription.
CoreSubject: One subject subscribed to over Core NATS.
CoreWildcard: A pattern subscribed to over Core NATS: orders.*, orders.>, >.
JetStreamMessage: Wrapper around an async_nats::jetstream::Message with ack semantics.
JetStreamOptions: What one JetStream publish states about itself, over and above its subject and payload.
JetStreamPublish: The JetStream publish policy: pure declaration, constructible anywhere.
JetStreamPublisher: The live JetStream publisher. Cheap to clone.
JetStreamSubject: A subject read through a JetStream pull consumer on one stream.
NatsBroker: A NATS broker: configuration captured, no I/O performed yet.
NatsPublish: The Core NATS publish policy: pure declaration, constructible anywhere.
NatsPublisher: The live Core NATS publisher. Cheap to clone.
NatsSubscriber: A NATS subscription.
NonZeroDuration: A Duration that is known not to be zero.
PublishAck: The acknowledgement a JetStream stream returns for an accepted publish. PublishAck is an acknowledgment received after successfully publishing a message.

Enums§

DeliverPolicy: DeliverPolicy determines how the consumer should select the first message to deliver.
NatsError: Errors surfaced by the NATS broker implementation.
NatsMessage: A NATS delivery. Two flavours: core NATS (no ack) and JetStream (real ack/nack/redelivery).

Constants§

PARTITION_KEY_HEADER: The well-known header key for per-message routing / partitioning.

Traits§

JetStreamPublishSteps: The per-message JetStream settings, as steps on the publish builder.
NatsPublishPolicy: A publish policy that pairs with a connected NATS broker without I/O.
NatsSubscription: A NATS subscription descriptor: CoreSubject, CoreWildcard or JetStreamSubject.