Struct pulsar::consumer::Consumer

pub struct Consumer<T: DeserializeMessage, Exe: Executor> { /* fields omitted */ }

the consumer is used to subscribe to a topic

use pulsar::{Consumer, SubType};
use futures::StreamExt;

let mut consumer: Consumer<TestData, _> = pulsar
    .consumer()
    .with_topic("non-persistent://public/default/test")
    .with_consumer_name("test_consumer")
    .with_subscription_type(SubType::Exclusive)
    .with_subscription("test_subscription")
    .build()
    .await?;

let mut counter = 0usize;
while let Some(Ok(msg)) = consumer.next().await {
    consumer.ack(&msg).await?;
    let data = match msg.deserialize() {
        Ok(data) => data,
        Err(e) => {
            log::error!("could not deserialize message: {:?}", e);
            break;
        }
    };

    counter += 1;
    log::info!("got {} messages", counter);
}

Implementations

impl<T: DeserializeMessage, Exe: Executor> Consumer<T, Exe>

pub fn builder(pulsar: &Pulsar<Exe>) -> ConsumerBuilder<Exe>

pub async fn check_connection<'_>(&'_ self) -> Result<(), Error>

pub async fn ack<'_, '_>(
    &'_ mut self,
    msg: &'_ Message<T>
) -> Result<(), ConsumerError>

pub async fn cumulative_ack<'_, '_>(
    &'_ mut self,
    msg: &'_ Message<T>
) -> Result<(), ConsumerError>

pub async fn nack<'_, '_>(
    &'_ mut self,
    msg: &'_ Message<T>
) -> Result<(), ConsumerError>

pub fn topics(&self) -> Vec<String>

pub fn connections(&self) -> Vec<&Url>

pub fn options(&self) -> &ConsumerOptions

pub fn dead_letter_policy(&self) -> Option<&DeadLetterPolicy>

pub fn subscription(&self) -> &str

pub fn sub_type(&self) -> SubType

pub fn batch_size(&self) -> Option<u32>

pub fn consumer_name(&self) -> Option<&str>

pub fn consumer_id(&self) -> Vec<u64>

pub fn unacked_message_redelivery_delay(&self) -> Option<Duration>

pub fn last_message_received(&self) -> Option<DateTime<Utc>>

pub fn messages_received(&self) -> u64

Trait Implementations

impl<T: DeserializeMessage + 'static, Exe: Executor> Stream for Consumer<T, Exe>

type Item = Result<Message<T>, Error>

Values yielded by the stream.

fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.