Struct async_nats::jetstream::stream::Stream

source · [−]

pub struct Stream {
    pub info: Info,
    /* private fields */
}

Expand description

Handle to operations that can be performed on a Stream.

Fields

info: Info

Implementations

source

impl Stream

source

pub async fn get_raw_message(&self, sequence: u64) -> Result<RawMessage, Error>

Get a raw message from the stream.

Examples

#[tokio::main]
use futures::StreamExt;
use futures::TryStreamExt;

let client = async_nats::connect("localhost:4222").await?;
let context = async_nats::jetstream::new(client);

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

let publish_ack = context.publish("events".to_string(), "data".into()).await?;
let raw_message = stream.get_raw_message(publish_ack.sequence).await?;
println!("Retreived raw message {:?}", raw_message);

source

pub async fn create_consumer<C: IntoConsumerConfig + FromConsumer>(
&self,
config: C
) -> Result<Consumer<C>, Error>

Create a new Durable or Ephemeral Consumer (if durable_name was not provided) and returns the info from the server about created Consumer

Examples

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

let stream = jetstream.get_stream("events").await?;
let info = stream.create_consumer(consumer::pull::Config {
    durable_name: Some("pull".to_string()),
    ..Default::default()
}).await?;

source

pub async fn consumer_info<T: AsRef<str>>(&self, name: T) -> Result<Info, Error>

Retrieve Info about Consumer from the server.

Examples

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

let stream = jetstream.get_stream("events").await?;
let info = stream.consumer_info("pull").await?;

source

pub async fn get_consumer<T: FromConsumer + IntoConsumerConfig>(
&self,
name: &str
) -> Result<Consumer<T>, Error>

Get Consumer from the the server. Consumer iterators can be used to retrieve Messages for a given Consumer.

Examples

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

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

source

pub async fn get_or_create_consumer<T: FromConsumer + IntoConsumerConfig>(
 &self,
 name: &str,
 config: T
) -> Result<Consumer<T>, Error>

Create a Consumer with the given configuration if it is not present on the server. Returns a handle to the Consumer.

Note: This does not validate if the Consumer on the server is compatible with the configuration passed in except Push/Pull compatibility.

Examples

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

let stream = jetstream.get_stream("events").await?;
let consumer = stream.get_or_create_consumer("pull", consumer::pull::Config {
    durable_name: Some("pull".to_string()),
    ..Default::default()
}).await?;

source

pub async fn delete_consumer(&self, name: &str) -> Result<DeleteStatus, Error>

Delete a Consumer from the server.

Examples

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

jetstream.get_stream("events").await?
    .delete_consumer("pull").await?;