Struct async_nats::jetstream::consumer::Consumer

source ·
pub struct Consumer<T: IntoConsumerConfig> { /* private fields */ }

Implementations§

source§

impl Consumer<Config>

source

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

Returns a stream of messages for Pull Consumer.

§Example
use futures::StreamExt;
use futures::TryStreamExt;

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

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

jetstream.publish("events", "data".into()).await?;

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

let mut messages = consumer.messages().await?.take(100);
while let Some(Ok(message)) = messages.next().await {
    println!("got message {:?}", message);
    message.ack().await?;
}
Ok(())
source

pub fn stream(&self) -> StreamBuilder<'_>

Enables customization of Stream by setting timeouts, heartbeats, maximum number of messages or bytes buffered.

§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)
    .max_bytes_per_batch(1024)
    .messages()
    .await?;

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

pub fn fetch(&self) -> FetchBuilder<'_>

Returns a batch of specified number of messages, or if there are less messages on the Stream than requested, returns all available messages.

§Example
use futures::StreamExt;
use futures::TryStreamExt;

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

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

jetstream.publish("events", "data".into()).await?;

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

for _ in 0..100 {
    jetstream.publish("events", "data".into()).await?;
}

let mut messages = consumer.fetch().max_messages(200).messages().await?;
// will finish after 100 messages, as that is the number of messages available on the
// stream.
while let Some(Ok(message)) = messages.next().await {
    println!("got message {:?}", message);
    message.ack().await?;
}
Ok(())
source

pub fn batch(&self) -> BatchBuilder<'_>

Returns a batch of specified number of messages unless timeout happens first.

§Example
use futures::StreamExt;
use futures::TryStreamExt;

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

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

jetstream.publish("events", "data".into()).await?;

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

let mut messages = consumer.batch().max_messages(100).messages().await?;
while let Some(Ok(message)) = messages.next().await {
    println!("got message {:?}", message);
    message.ack().await?;
}
Ok(())
source

pub fn sequence(&self, batch: usize) -> Result<Sequence, BatchError>

Returns a sequence of Batches allowing for iterating over batches, and then over messages in those batches.

§Example
use futures::StreamExt;
use futures::TryStreamExt;

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

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

jetstream.publish("events", "data".into()).await?;

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

let mut iter = consumer.sequence(50).unwrap().take(10);
while let Ok(Some(mut batch)) = iter.try_next().await {
    while let Ok(Some(message)) = batch.try_next().await {
        println!("message received: {:?}", message);
    }
}
Ok(())
source§

impl Consumer<OrderedConfig>

source

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

Returns a stream of messages for Ordered Pull Consumer.

Ordered consumers uses single replica ephemeral consumer, no matter the replication factor of the Stream. It does not use acks, instead it tracks sequences and recreate itself whenever it sees mismatch.

§Example
use futures::StreamExt;
use futures::TryStreamExt;

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

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

jetstream.publish("events", "data".into()).await?;

let consumer = stream
    .get_or_create_consumer(
        "consumer",
        async_nats::jetstream::consumer::pull::OrderedConfig {
            name: Some("consumer".to_string()),
            ..Default::default()
        },
    )
    .await?;

let mut messages = consumer.messages().await?.take(100);
while let Some(Ok(message)) = messages.next().await {
    println!("got message {:?}", message);
}
Ok(())
source§

impl Consumer<Config>

source

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

Returns a stream of messages for Push Consumer.

§Example
use async_nats::jetstream::consumer::PushConsumer;
use futures::StreamExt;
use futures::TryStreamExt;

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

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

jetstream.publish("events", "data".into()).await?;

let consumer: PushConsumer = stream
    .get_or_create_consumer(
        "consumer",
        async_nats::jetstream::consumer::push::Config {
            durable_name: Some("consumer".to_string()),
            deliver_subject: "deliver".to_string(),
            ..Default::default()
        },
    )
    .await?;

let mut messages = consumer.messages().await?.take(100);
while let Some(Ok(message)) = messages.next().await {
    println!("got message {:?}", message);
    message.ack().await?;
}
Ok(())
source§

impl Consumer<OrderedConfig>

source

pub async fn messages<'a>(self) -> Result<Ordered, StreamError>

source§

impl<T: IntoConsumerConfig> Consumer<T>

source

pub fn new(config: T, info: Info, context: Context) -> Self

source§

impl<T: IntoConsumerConfig> Consumer<T>

source

pub async fn info(&mut self) -> Result<&Info, RequestError>

Retrieves info about Consumer from the server, updates the cached info inside Consumer and returns it.

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

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

let info = consumer.info().await?;
source

pub fn cached_info(&self) -> &Info

Returns cached Info for the Consumer. Cache is either from initial creation/retrieval of the Consumer or last call to Info.

§Examples
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 info = consumer.cached_info();

Trait Implementations§

source§

impl<T: Clone + IntoConsumerConfig> Clone for Consumer<T>

source§

fn clone(&self) -> Consumer<T>

Returns a copy of the value. Read more
1.0.0 · source§

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source. Read more
source§

impl<T: Debug + IntoConsumerConfig> Debug for Consumer<T>

source§

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter. Read more

Auto Trait Implementations§

§

impl<T> Freeze for Consumer<T>
where T: Freeze,

§

impl<T> !RefUnwindSafe for Consumer<T>

§

impl<T> Send for Consumer<T>
where T: Send,

§

impl<T> Sync for Consumer<T>
where T: Sync,

§

impl<T> Unpin for Consumer<T>
where T: Unpin,

§

impl<T> !UnwindSafe for Consumer<T>

Blanket Implementations§

source§

impl<T> Any for T
where T: 'static + ?Sized,

source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
source§

impl<T> Borrow<T> for T
where T: ?Sized,

source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
source§

impl<T> CloneToUninit for T
where T: Clone,

source§

unsafe fn clone_to_uninit(&self, dst: *mut T)

🔬This is a nightly-only experimental API. (clone_to_uninit)
Performs copy-assignment from self to dst. Read more
source§

impl<T> From<T> for T

source§

fn from(t: T) -> T

Returns the argument unchanged.

source§

impl<T> Instrument for T

source§

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
source§

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
source§

impl<T, U> Into<U> for T
where U: From<T>,

source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

source§

impl<T> Same for T

source§

type Output = T

Should always be Self
source§

impl<T> ToOwned for T
where T: Clone,

source§

type Owned = T

The resulting type after obtaining ownership.
source§

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. Read more
source§

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. Read more
source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

source§

type Error = Infallible

The type returned in the event of a conversion error.
source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

source§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
source§

impl<V, T> VZip<V> for T
where V: MultiLane<T>,

source§

fn vzip(self) -> V

source§

impl<T> WithSubscriber for T

source§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
source§

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more