lapin

You can configure lapin to automatically reconnect and replay exchanges, queues, bindings, and consumers after a network failure by calling .enable_auto_recover() on ConnectionProperties.

If you encounter a recoverable error on a channel, you can block until the recovery process is complete by calling channel.wait_for_recovery(error).await.

use lapin::ConnectionProperties;

let props = ConnectionProperties::default().enable_auto_recover();

If you prefer a callback-based approach over polling the Consumer stream, you can implement the ConsumerDelegate trait. This allows you to have deliveries dispatched automatically.

To use it, implement the trait and pass your instance to Consumer::set_delegate. This method enables parallel handling of messages by spawning the delegate on the executor for each message.

Trait Methods

• on_new_delivery(&self, delivery: DeliveryResult) -> Pin<Box<dyn Future<Output = ()> + Send>>: Called for each new delivery, cancellation, or error from the server.
• drop_prefetched_messages(&self) -> Pin<Box<dyn Future<Output = ()> + Send>>: (Optional) Called when the consumer is asked to discard buffered messages (e.g., on channel close).

The Channel struct is the primary entry point for most AMQP operations, such as declaring exchanges, publishing messages, and consuming from queues. While the AMQP specification suggests one channel per OS thread, Channel in lapin is Clone + Send + Sync. This means you can safely clone a Channel and use the clones across different asynchronous tasks or OS threads.

Channels are typically obtained from a Connection using Connection::create_channel.

The can_be_recovered() method indicates whether the error is of a type that the automatic recovery logic can attempt to handle. This is used internally by the auto-recovery logic and requires ConnectionProperties::enable_auto_recover to be set.

Recoverable errors:

• InvalidChannel
• InvalidChannelState
• InvalidConnectionState
• IOError
• ProtocolError
• MissingHeartbeatError

Non-recoverable errors:

• ChannelsLimitReached
• InvalidProtocolVersion
• RuntimeShutdownError
• ParsingError
• SerialisationError
• AuthProviderError
• FutureCompleted
• NoDefaultRuntime

When you call Channel::basic_publish, it returns a PublisherConfirm future. This future resolves to a Confirmation enum, which indicates whether the broker successfully received and processed the message.

Important Lifecycle Note: If you drop a PublisherConfirm without awaiting it, the library automatically registers it with Channel::wait_for_confirms to ensure that confirmations are not silently lost. This prevents a common race condition where a developer might assume a message was sent successfully simply because they didn't encounter an error during the publish call.

Lapin is runtime-agnostic. You must pick exactly one async runtime feature flag during installation:

The ConnectionState enum represents the various stages of an AMQP connection's lifecycle. Monitoring these states allows you to implement logic for handling reconnections, errors, or graceful shutdowns.

Connection States

• Initial: The connection object is created, but the TCP handshake hasn't started.
• Connecting: The TCP connection is open and the AMQP handshake is in progress.
• Connected: The AMQP handshake completed successfully; the connection is ready for use.
• Closing: A Connection.Close frame has been sent; waiting for Connection.Close-Ok.
• Closed: The connection has been closed normally.
• Reconnecting: The connection was lost and is being re-established via automatic recovery.
• Error: The connection was closed due to a protocol or IO error.

You must pick at most one TLS backend feature flag:

The lapin library follows the standard AMQP hierarchy:

1. Connection: A single TCP socket to the broker. Typically, one process creates one connection and reuses it.
2. Channel: A lightweight virtual connection multiplexed over a Connection. All AMQP operations (declaring queues, publishing, consuming, etc.) are performed through channels. They are cheap to open.
3. Consumer: An async Stream of message::Delivery values obtained by calling Channel::basic_consume. Each delivery must be explicitly acknowledged (e.g., via delivery.ack(...)) once processed.
4. PublisherConfirm: A future returned by Channel::basic_publish that resolves to a Confirmation once the broker has acknowledged the message. This requires calling Channel::confirm_select first.

When the rustls feature is active, you can choose a certificate store via these flags:

When using rustls, you must enable at least one crypto provider: