Struct StreamBuilder

pub struct StreamBuilder<'a> { /* private fields */ }

Used for building configuration for a Stream. Created by a Consumer::stream on a Consumer.

Examples

use futures::StreamExt;
use async_nats::jetstream::consumer::PullConsumer;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: PullConsumer = jetstream
    .get_stream("events").await?
    .get_consumer("pull").await?;

let mut messages = consumer.stream()
    .max_messages_per_batch(100)
    .max_bytes_per_batch(1024)
    .messages().await?;

while let Some(message) = messages.next().await {
    let message = message?;
    println!("message: {:?}", message);
    message.ack().await?;
}

Implementations

impl<'a> StreamBuilder<'a>

pub fn new(consumer: &'a Consumer<Config>) -> Self

pub fn max_bytes_per_batch(self, max_bytes: usize) -> Self

Sets max bytes that can be buffered on the Client while processing already received messages. Higher values will yield better performance, but also potentially increase memory usage if application is acknowledging messages much slower than they arrive.

Default values should provide reasonable balance between performance and memory usage.

Examples

use async_nats::jetstream::consumer::PullConsumer;
use futures::StreamExt;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: PullConsumer = jetstream
    .get_stream("events")
    .await?
    .get_consumer("pull")
    .await?;

let mut messages = consumer
    .stream()
    .max_bytes_per_batch(1024)
    .messages()
    .await?;

while let Some(message) = messages.next().await {
    let message = message?;
    println!("message: {:?}", message);
    message.ack().await?;
}

pub fn max_messages_per_batch(self, batch: usize) -> Self

Sets max number of messages that can be buffered on the Client while processing already received messages. Higher values will yield better performance, but also potentially increase memory usage if application is acknowledging messages much slower than they arrive.

Default values should provide reasonable balance between performance and memory usage.

Examples

use async_nats::jetstream::consumer::PullConsumer;
use futures::StreamExt;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: PullConsumer = jetstream
    .get_stream("events")
    .await?
    .get_consumer("pull")
    .await?;

let mut messages = consumer
    .stream()
    .max_messages_per_batch(100)
    .messages()
    .await?;

while let Some(message) = messages.next().await {
    let message = message?;
    println!("message: {:?}", message);
    message.ack().await?;
}

pub fn heartbeat(self, heartbeat: Duration) -> Self

Sets heartbeat which will be send by the server if there are no messages for a given Consumer pending.

Examples

use async_nats::jetstream::consumer::PullConsumer;
use futures::StreamExt;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: PullConsumer = jetstream
    .get_stream("events")
    .await?
    .get_consumer("pull")
    .await?;

let mut messages = consumer
    .stream()
    .heartbeat(std::time::Duration::from_secs(10))
    .messages()
    .await?;

while let Some(message) = messages.next().await {
    let message = message?;
    println!("message: {:?}", message);
    message.ack().await?;
}

pub fn expires(self, expires: Duration) -> Self

Low level API that does not need tweaking for most use cases. Sets how long each batch request waits for whole batch of messages before timing out. Consumer pending.

Examples

use async_nats::jetstream::consumer::PullConsumer;
use futures::StreamExt;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: PullConsumer = jetstream
    .get_stream("events")
    .await?
    .get_consumer("pull")
    .await?;

let mut messages = consumer
    .stream()
    .expires(std::time::Duration::from_secs(30))
    .messages()
    .await?;

while let Some(message) = messages.next().await {
    let message = message?;
    println!("message: {:?}", message);
    message.ack().await?;
}

pub async fn messages(self) -> Result<Stream, StreamError>

Creates actual Stream with provided configuration.

Examples

use async_nats::jetstream::consumer::PullConsumer;
use futures::StreamExt;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: PullConsumer = jetstream
    .get_stream("events")
    .await?
    .get_consumer("pull")
    .await?;

let mut messages = consumer
    .stream()
    .max_messages_per_batch(100)
    .messages()
    .await?;

while let Some(message) = messages.next().await {
    let message = message?;
    println!("message: {:?}", message);
    message.ack().await?;
}