Module jetstream

JetStream is a built-in persistence layer for NATS that provides powerful stream-based messaging capabilities, with integrated support for both at least once and exactly once delivery semantics.

To begin using JetStream, you need to create a new Context object, which serves as the entry point to the JetStream API.

Examples

Below are some examples that demonstrate how to use JetStream for publishing and consuming messages.

Publishing and Consuming Messages

This example demonstrates how to publish messages to a JetStream stream and consume them using a pull-based consumer.

use futures::StreamExt;
use futures::TryStreamExt;

// Connect to NATS server
let client = async_nats::connect("localhost:4222").await?;

// Create a JetStream instance
let jetstream = async_nats::jetstream::new(client);

// Get or create a stream
let stream = jetstream
    .get_or_create_stream(async_nats::jetstream::stream::Config {
        name: "events".to_string(),
        max_messages: 10_000,
        ..Default::default()
    })
    .await?;

// Publish a message to the stream
jetstream.publish("events", "data".into()).await?;

// Get or create a pull-based consumer
let consumer = stream
    .get_or_create_consumer(
        "consumer",
        async_nats::jetstream::consumer::pull::Config {
            durable_name: Some("consumer".to_string()),
            ..Default::default()
        },
    )
    .await?;

// Consume messages from the consumer
let mut messages = consumer.messages().await?.take(100);
while let Ok(Some(message)) = messages.try_next().await {
    println!("message receiver: {:?}", message);
    message.ack().await?;
}

Ok(())

Consuming Messages in Batches

This example demonstrates how to consume messages in batches from a JetStream stream using a sequence-based consumer.

use futures::StreamExt;
use futures::TryStreamExt;

// Connect to NATS server
let client = async_nats::connect("localhost:4222").await?;

// Create a JetStream instance
let jetstream = async_nats::jetstream::new(client);

// Get or create a stream
let stream = jetstream
    .get_or_create_stream(async_nats::jetstream::stream::Config {
        name: "events".to_string(),
        max_messages: 10_000,
        ..Default::default()
    })
    .await?;

// Publish a message to the stream
jetstream.publish("events", "data".into()).await?;

// Get or create a pull-based consumer
let consumer = stream
    .get_or_create_consumer(
        "consumer",
        async_nats::jetstream::consumer::pull::Config {
            durable_name: Some("consumer".to_string()),
            ..Default::default()
        },
    )
    .await?;

// Consume messages from the consumer in batches
let mut batches = consumer.sequence(50)?.take(10);
while let Ok(Some(mut batch)) = batches.try_next().await {
    while let Some(Ok(message)) = batch.next().await {
        println!("message receiver: {:?}", message);
        message.ack().await?;
    }
}

Ok(())

Re-exports

• pub use context::Context;
• pub use message::AckKind;
• pub use message::Message;

Modules

• account
• consumer
  Push and Pull Consumer API.
• context
  Manage operations on Context, create/delete/update Stream
• kv
  A Key-Value store built on top of JetStream, allowing you to store and retrieve data using simple key-value pairs.
• message
  A wrapped crate::Message with JetStream related methods.
• object_store
  Object Store module
• publish
  Publish JetStream messages.
• response
  A low level JetStream responses.
• stream
  Manage operations on a Stream, create/delete/update Consumer.

Structs

• Error
  Error type returned from an API response when an error occurs.
• ErrorCode

Functions

• new
  Creates a new JetStream Context that provides JetStream API for managing and using Streams, Consumers, key value and object store.
• with_domain
  Creates a new JetStream Context with given JetStream domain.
• with_prefix
  Creates a new JetStream Context with given JetStream prefix. By default it is $JS.API.