Struct PullConsumer 

pub struct PullConsumer<T, C: CodecType> { /* private fields */ }

A typed pull consumer with configurable codec.

Type Parameters

• T - The message type
• C - The codec type used for deserialization

Implementations

impl<T, C: CodecType> PullConsumer<T, C>

pub fn inner(&self) -> &Consumer<Config>

Get the underlying async-nats consumer.

pub fn name(&self) -> &str

Get consumer name.

impl<T: DeserializeOwned, C: CodecType> PullConsumer<T, C>

pub async fn fetch(&self, batch_size: usize) -> Result<PullBatch<T, C>>

Fetch a batch of messages.

Example

use intercom::{Client, MsgPackCodec};
use serde::{Deserialize, Serialize};
use futures::StreamExt;

#[derive(Serialize, Deserialize, Debug)]
struct Event { id: u64 }

let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
let jetstream = client.jetstream();
let stream = jetstream.get_stream("events").await?;
let consumer = stream.get_pull_consumer::<Event>("my-consumer").await?;

let mut messages = consumer.fetch(10).await?;
while let Some(result) = messages.next().await {
    let msg = result?;
    println!("Got: {:?}", msg.payload);
    msg.ack().await?;
}

pub fn fetch_builder(&self) -> FetchBuilder<T, C>

Fetch messages with a builder for advanced options.

pub async fn messages(&self) -> Result<PullMessages<T, C>>

Create a continuous message stream.

Returns a Stream that continuously delivers messages.

Example

use intercom::{Client, MsgPackCodec};
use serde::{Deserialize, Serialize};
use futures::StreamExt;

#[derive(Serialize, Deserialize, Debug)]
struct Event { id: u64 }

let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
let jetstream = client.jetstream();
let stream = jetstream.get_stream("events").await?;
let consumer = stream.get_pull_consumer::<Event>("my-consumer").await?;

let mut messages = consumer.messages().await?;
while let Some(result) = messages.next().await {
    let msg = result?;
    println!("Got: {:?}", msg.payload);
    msg.ack().await?;
}