Struct lapin::Consumer

pub struct Consumer { /* fields omitted */ }

Continuously consumes message from a Queue.

A consumer represents a stream of messages created from the basic.consume AMQP command. It continuously receives messages from the queue, as opposed to the basic.get command, which retrieves only a single message.

A consumer is obtained by calling Channel::basic_consume with the queue name.

New messages from this consumer can be accessed by obtaining the iterator from the consumer. This iterator returns new messages and the associated channel in the form of a DeliveryResult for as long as the consumer is subscribed to the queue.

It is also possible to set a delegate to be spawned via set_delegate.

Message acknowledgment

There are two ways for acknowledging a message:

• If the flag BasicConsumeOptions::no_ack is set to true while obtaining the consumer from Channel::basic_consume, the server implicitely acknowledges each message after it has been sent.
• If the flag BasicConsumeOptions::no_ack is set to false, a message has to be explicitely acknowledged or rejected with Channel::basic_ack, Channel::basic_reject or Channel::basic_nack. See the documentation at Delivery for further information.

Also see the RabbitMQ documentation about Acknowledgement Modes.

Consumer Prefetch

To limit the maximum number of unacknowledged messages arriving, you can call Channel::basic_qos before creating the consumer.

Also see the RabbitMQ documentation about Consumer Prefetch.

Cancel subscription

To stop receiving messages, call Channel::basic_cancel with the consumer tag of this consumer.

Example

use lapin::{options::*, types::FieldTable, Connection, ConnectionProperties, Result};
use futures_executor::LocalPool;
use futures_util::stream::StreamExt;
use std::future::Future;

let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());

let res: Result<()> = LocalPool::new().run_until(async {
    let conn = Connection::connect(
        &addr,
        ConnectionProperties::default().with_default_executor(8),
    )
    .await?;
    let channel = conn.create_channel().await?;
    let mut consumer = channel
        .basic_consume(
            "hello",
            "my_consumer",
            BasicConsumeOptions::default(),
            FieldTable::default(),
        )
        .await?;

    while let Some(delivery) = consumer.next().await {
        let (channel, delivery) = delivery.expect("error in consumer");
        channel
            .basic_ack(delivery.delivery_tag, BasicAckOptions::default())
            .await?;
    }
    Ok(())
});

Implementations

impl Consumer

pub fn tag(&self) -> ShortString

Gets the consumer tag.

If no consumer tag was specified when obtaining the consumer from the channel, this contains the server generated consumer tag.

pub fn set_delegate<D: ConsumerDelegate + 'static>(
    &self,
    delegate: D
) -> Result<()>

Automatically spawns the delegate on the executor for each message.

Enables parallel handling of the messages.

Trait Implementations

impl Clone for Consumer

fn clone(&self) -> Consumer

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for Consumer

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

Formats the value using the given formatter.

impl IntoIterator for Consumer

type Item = Result<(Channel, Delivery)>

The type of the elements being iterated over.

type IntoIter = ConsumerIterator

Which kind of iterator are we turning this into?

fn into_iter(self) -> Self::IntoIter

Creates an iterator from a value.

impl Stream for Consumer

type Item = Result<(Channel, Delivery)>

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.