Struct RabbitMqBackend 

pub struct RabbitMqBackend { /* private fields */ }

A Backend backed by a single RabbitMQ connection.

• One Connection.
• One publishing Channel in confirm mode, shared behind a tokio::sync::Mutex. Every publish (enqueue, retry, defer, dead-letter) is mandatory and waits for the broker’s confirmation. The channel is reopened lazily if a channel exception closed it. Nothing is ever declared on it.
• One long-lived Channel for the hold queue declarations defer makes on demand. A declaration is the one thing the broker routinely refuses (PRECONDITION_FAILED closes the channel it ran on), so it is kept away from the publishes it would otherwise take down with it.
• One fresh Channel per consume call, so each consumer gets its own basic_qos prefetch window and a failure on one consumer cannot take down the others.
• One short-lived channel per declare call, so a rejected declaration (e.g. re-declaring an existing queue with different arguments, which RabbitMQ answers with PRECONDITION_FAILED and closes the channel) cannot poison the other channels.

Reconnection is out of scope: when the connection is lost, consumer streams end and subsequent operations fail.

See topology for the queues this creates.

Implementations

impl RabbitMqBackend

pub async fn connect(uri: &str) -> Result<Self>

Connect to uri with RabbitMqOptions::default.

use queuey_rabbitmq::RabbitMqBackend;

let backend = RabbitMqBackend::connect("amqp://guest:guest@localhost:5672/%2f").await?;

pub async fn with_options(uri: &str, options: RabbitMqOptions) -> Result<Self>

Connect to uri with explicit options.

pub fn options(&self) -> &RabbitMqOptions

The options this backend was built with.

pub fn retry_queue_name(&self, queue: &str) -> String

The name of the retry (wait) queue backing queue.

pub fn dead_queue_name(&self, queue: &str) -> String

The name of the dead-letter queue backing queue.

pub fn deferred_queue_name(&self, queue: &str, ttl_ms: u32) -> String

The name of the hold queue that ttl_ms-long deferrals of queue wait in.

There is one per distinct rounded delay, and it is created on demand by defer rather than by declare; see topology.

Trait Implementations

impl Backend for RabbitMqBackend

fn defer<'life0, 'life1, 'async_trait>( &'life0 self, envelope: &'life1 Envelope, delay: Duration, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Hold envelope for delay, then put it back on its own queue.

The queue must have been declared through this backend

Deferring onto a queue this backend instance never declared is Error::UnknownQueue, not a best-effort publish. A hold queue has to know its main queue’s durability and dead-letter it back by name, and neither can be guessed: a transient hold queue in front of a durable queue loses jobs on a restart, and a TTL expiry into a queue that does not exist is dropped by the broker in silence. Unlike a mandatory publish, nothing is returned and nothing is reported.

Producer::new and WorkerBuilder::build declare the whole queue set, so anything built through them can defer. Producer::new_undeclared deliberately does not, so a producer built that way can enqueue but cannot defer until something in the process declares the queue.

The delay has a ceiling

Delays are rounded up to deferred_granularity and capped at MAX_DEFERRAL_MS (~24.8 days); a longer one is an error rather than a shorter wait.

fn declare<'life0, 'life1, 'async_trait>( &'life0 self, queues: &'life1 [QueueConfig], ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Idempotently create all queues (plus any retry / dead-letter infrastructure).

fn publish<'life0, 'life1, 'async_trait>( &'life0 self, envelope: &'life1 Envelope, delay: Option<Duration>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Publish envelope to envelope.queue, optionally delayed.

fn consume<'life0, 'life1, 'async_trait>( &'life0 self, queue: &'life1 QueueConfig, ) -> Pin<Box<dyn Future<Output = Result<DeliveryStream>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Start consuming queue with the given prefetch. The stream ends when the backend is closed or the connection is lost.

fn close<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Graceful shutdown: stop all consumers, flush, close connections.

impl Debug for RabbitMqBackend

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

Formats the value using the given formatter.