### Full AMQP Example with Tokio

Source: https://github.com/amqp-rs/lapin/blob/main/README.md

A comprehensive example demonstrating connection, channel creation, queue declaration, basic publishing, and consumption using Tokio runtime. Ensure AMQP_ADDR environment variable is set or defaults to amqp://127.0.0.1:5672/%2f. RUST_LOG defaults to info.

```rust
use async_rs::{Runtime, traits::*};
use futures_lite::stream::StreamExt;
use lapin::{
    BasicProperties,
    Confirmation,
    Connection,
    ConnectionProperties,
    Result,
    options::*,
    types::FieldTable,
};
use tracing::info;

#[tokio::main]
async fn main() -> Result<()> {
    if std::env::var("RUST_LOG").is_err() {
        unsafe { std::env::set_var("RUST_LOG", "info") };
    }

    tracing_subscriber::fmt::init();

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

    let conn = Connection::connect_with_runtime(
        &addr,
        ConnectionProperties::default().with_connection_name("pubsub-example".into()),
        runtime.clone(),
    )
    .await?;

    info!("CONNECTED");

    let channel_a = conn.create_channel().await?;
    let channel_b = conn.create_channel().await?;

    let queue = channel_a
        .queue_declare(
            "hello".into(),
            QueueDeclareOptions::default(),
            FieldTable::default(),
        )
        .await?;

    info!(?queue, "Declared queue");

    let mut consumer = channel_b
        .basic_consume(
            "hello".into(),
            "my_consumer".into(),
            BasicConsumeOptions::default(),
            FieldTable::default(),
        )
        .await?;
    let cons = runtime.spawn(async move {
        info!("will consume");
        while let Some(delivery) = consumer.next().await {
            let delivery = delivery?;
            delivery.ack(BasicAckOptions::default()).await?;
        }
        Ok(())
    });

    let payload = b"Hello world!";

    for _ in 0..1500000 {
        let confirm = channel_a
            .basic_publish(
                "".into(),
                "hello".into(),
                BasicPublishOptions::default(),
                payload,
                BasicProperties::default(),
            )
            .await? 
            .await?;
        assert_eq!(confirm, Confirmation::NotRequested);
    }

    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    channel_b
        .basic_cancel("my_consumer".into(), BasicCancelOptions::default())
        .await?;
    cons.await
}
```

--------------------------------

### Connect to AMQP Server with Default Runtime

Source: https://context7.com/amqp-rs/lapin/llms.txt

Establishes a TCP connection to an AMQP server using a URI string. The URI must be URL-encoded for the virtual host. This is the simplest connection method for the default runtime.

```rust
use lapin::{Connection, ConnectionProperties, Result};

#[tokio::main]
async fn main() -> Result<()> {
    // Connect to RabbitMQ with default virtual host "/"
    let conn = Connection::connect(
        "amqp://guest:guest@127.0.0.1:5672/%2f",
        ConnectionProperties::default(),
    )
    .await?;

    println!("Connected! Status: {:?}", conn.status().state());

    // Close the connection gracefully
    conn.close(200, "Normal shutdown".into()).await?;
    Ok(())
}
```

--------------------------------

### Connection::connect

Source: https://context7.com/amqp-rs/lapin/llms.txt

Establishes a TCP connection to an AMQP server using a URI string. Supports default async runtimes.

```APIDOC
## POST /connect

### Description
Establishes a TCP connection to an AMQP server using a URI string. The URI format supports specifying the host, port, credentials, and virtual host. The virtual host must be URL-encoded (e.g., `%2f` for `/`). This is the simplest way to connect when using the default runtime.

### Method
POST

### Endpoint
/connect

### Parameters
#### Query Parameters
- **uri** (string) - Required - The AMQP server URI (e.g., `amqp://guest:guest@127.0.0.1:5672/%2f`).
- **properties** (ConnectionProperties) - Required - Default connection properties.

### Request Example
```rust
use lapin::{Connection, ConnectionProperties, Result};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect(
        "amqp://guest:guest@127.0.0.1:5672/%2f",
        ConnectionProperties::default(),
    )
    .await?;

    println!("Connected! Status: {:?}", conn.status().state());
    conn.close(200, "Normal shutdown".into()).await?;
    Ok(())
}
```

### Response
#### Success Response (200)
- **connection** (Connection) - An established AMQP connection object.

#### Response Example
```rust
// Successful connection returns a Connection object
```
```

--------------------------------

### Create and Manage AMQP Channels

Source: https://context7.com/amqp-rs/lapin/llms.txt

Demonstrates creating multiple channels on a single connection for different purposes, such as publishing and consuming. Channels are lightweight and share the underlying TCP connection.

```rust
use lapin::{Connection, ConnectionProperties, Result};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect(
        "amqp://127.0.0.1:5672/%2f",
        ConnectionProperties::default(),
    )
    .await?;

    // Create channels for different purposes
    let publish_channel = conn.create_channel().await?;
    let consume_channel = conn.create_channel().await?;

    println!("Publisher channel ID: {}", publish_channel.id());
    println!("Consumer channel ID: {}", consume_channel.id());

    // Close channels explicitly (optional - they close when dropped)
    publish_channel.close(200, "Done publishing".into()).await?;
    consume_channel.close(200, "Done consuming".into()).await?;

    Ok(())
}
```

--------------------------------

### Connect to AMQP Server with Explicit Runtime

Source: https://context7.com/amqp-rs/lapin/llms.txt

Connects to an AMQP server specifying an explicit async runtime. Recommended for applications needing explicit runtime management or using non-default runtimes like smol.

```rust
use async_rs::{Runtime, traits::*};
use lapin::{Connection, ConnectionProperties, Result};

#[tokio::main]
async fn main() -> Result<()> {
    let runtime = Runtime::tokio_current();

    let conn = Connection::connect_with_runtime(
        "amqp://127.0.0.1:5672/%2f",
        ConnectionProperties::default()
            .with_connection_name("my-application".into()),
        runtime,
    )
    .await?;

    println!("Connected with custom runtime!");
    Ok(())
}
```

--------------------------------

### Connection::connect_with_runtime

Source: https://context7.com/amqp-rs/lapin/llms.txt

Connects to an AMQP server with an explicit runtime specification, offering more control over the async runtime.

```APIDOC
## POST /connect_with_runtime

### Description
Connects to an AMQP server with an explicit runtime specification. This method provides more control over which async runtime to use and is recommended for applications that need explicit runtime management or want to use non-default runtimes like smol.

### Method
POST

### Endpoint
/connect_with_runtime

### Parameters
#### Query Parameters
- **uri** (string) - Required - The AMQP server URI (e.g., `amqp://127.0.0.1:5672/%2f`).
- **properties** (ConnectionProperties) - Required - Connection properties, can be customized.
- **runtime** (Runtime) - Required - The async runtime to use (e.g., `Runtime::tokio_current()`).

### Request Example
```rust
use async_rs::{Runtime, traits::*};
use lapin::{Connection, ConnectionProperties, Result};

#[tokio::main]
async fn main() -> Result<()> {
    let runtime = Runtime::tokio_current();

    let conn = Connection::connect_with_runtime(
        "amqp://127.0.0.1:5672/%2f",
        ConnectionProperties::default()
            .with_connection_name("my-application".into()),
        runtime,
    )
    .await?;

    println!("Connected with custom runtime!");
    Ok(())
}
```

### Response
#### Success Response (200)
- **connection** (Connection) - An established AMQP connection object.

#### Response Example
```rust
// Successful connection returns a Connection object
```
```

--------------------------------

### Create and Process AMQP Messages with basic_consume

Source: https://context7.com/amqp-rs/lapin/llms.txt

Creates a consumer to continuously receive messages from a queue. Each message requires manual acknowledgment. Ensure the queue is declared before consuming.

```rust
use futures_lite::StreamExt;
use lapin::{Connection, ConnectionProperties, Result, options::*,
 types::FieldTable};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    // Ensure queue exists
    channel.queue_declare("tasks".into(), QueueDeclareOptions::default(), FieldTable::default()).await?;

    // Create a consumer
    let mut consumer = channel
        .basic_consume(
            "tasks".into(),           // Queue name
            "my-consumer".into(),     // Consumer tag (unique identifier)
            BasicConsumeOptions {
                no_local: false,      // Receive messages from same connection
                no_ack: false,        // Manual acknowledgment required
                exclusive: false,     // Allow other consumers
                nowait: false,        // Wait for server confirmation
            },
            FieldTable::default(),
        )
        .await?;

    println!("Consumer started, tag: {}", consumer.tag());

    // Process messages
    while let Some(delivery_result) = consumer.next().await {
        match delivery_result {
            Ok(delivery) => {
                let data = String::from_utf8_lossy(&delivery.data);
                println!(
                    "Received message: {} (delivery_tag: {}, exchange: {}, routing_key: {})",
                    data,
                    delivery.delivery_tag,
                    delivery.exchange,
                    delivery.routing_key
                );

                // Acknowledge successful processing
                delivery.ack(BasicAckOptions::default()).await?;
            }
            Err(error) => {
                eprintln!("Consumer error: {:?}", error);
                break;
            }
        }
    }

    Ok(())
}
```

--------------------------------

### Listen for Connection Events in Lapin

Source: https://context7.com/amqp-rs/lapin/llms.txt

Returns a stream of connection events for monitoring connection state changes. This is useful for logging, metrics, and implementing custom recovery logic.

```rust
use futures_lite::StreamExt;
use lapin::{Connection, ConnectionProperties, Event, Result};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;

    // Get events stream
    let mut events = conn.events_listener();

    // Spawn event handler
    tokio::spawn(async move {
        while let Some(event) = events.next().await {
            match event {
                Event::Connected => println!("Event: Connected to broker"),
                Event::ConnectionBlocked(reason) => println!("Event: Connection blocked: {}", reason),
                Event::ConnectionUnblocked => println!("Event: Connection unblocked"),
                Event::Error(error) => println!("Event: Error occurred: {:?}", error),
                _ => println!("Event: {:?}", event),
            }
        }
    });

    // Continue with normal operations
    let channel = conn.create_channel().await?;
    println!("Channel created: {}", channel.id());

    // Keep running to observe events
    tokio::time::sleep(std::time::Duration::from_secs(60)).await;
    Ok(())
}
```

--------------------------------

### Configure Connection Properties with Auto-Recovery

Source: https://context7.com/amqp-rs/lapin/llms.txt

Configures detailed connection properties including connection name, client properties, locale, and automatic recovery with a custom backoff strategy. Use this to customize connection behavior and resilience.

```rust
use lapin::{Connection, ConnectionProperties, Result};
use backon::ExponentialBuilder;

#[tokio::main]
async fn main() -> Result<()> {
    let properties = ConnectionProperties::default()
        // Set a descriptive connection name visible in RabbitMQ management
        .with_connection_name("order-processor".into())
        // Set custom client properties
        .with_client_property("application".into(), "order-service".into())
        // Configure locale
        .with_locale("en_US".into())
        // Enable automatic connection recovery
        .enable_auto_recover()
        // Configure custom backoff strategy
        .configure_backoff(|builder| {
            builder
                .with_max_times(10)
                .with_min_delay(std::time::Duration::from_millis(100))
                .with_max_delay(std::time::Duration::from_secs(30))
        });

    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", properties).await?;
    println!("Connected with auto-recovery enabled");
    Ok(())
}
```

--------------------------------

### Publish Message with Basic Properties

Source: https://context7.com/amqp-rs/lapin/llms.txt

Publishes a message to an exchange with optional properties like content type, delivery mode, priority, message ID, and correlation ID. Ensure the queue exists before publishing.

```rust
use lapin::{Connection, ConnectionProperties, BasicProperties, Result, options::*, types::FieldTable};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    // Ensure queue exists
    channel.queue_declare("tasks".into(), QueueDeclareOptions::default(), FieldTable::default()).await?;

    // Simple publish to default exchange (routing key = queue name)
    let payload = b"Hello, World!";
    channel
        .basic_publish(
            "".into(),           // Default exchange
            "tasks".into(),      // Routing key (queue name for default exchange)
            BasicPublishOptions::default(),
            payload,
            BasicProperties::default(),
        )
        .await?;

    // Publish with message properties
    let order_data = r#"{"order_id": 12345, "amount": 99.99}"#;
    channel
        .basic_publish(
            "".into(),
            "tasks".into(),
            BasicPublishOptions::default(),
            order_data.as_bytes(),
            BasicProperties::default()
                .with_content_type("application/json".into())
                .with_delivery_mode(2)  // Persistent
                .with_priority(5)
                .with_message_id("msg-12345".into())
                .with_correlation_id("req-67890".into()),
        )
        .await?;

    // Publish with mandatory flag (returns message if unroutable)
    channel
        .basic_publish(
            "".into(),
            "tasks".into(),
            BasicPublishOptions {
                mandatory: true,  // Return message if no queue is bound
                immediate: false,
            },
            payload,
            BasicProperties::default(),
        )
        .await?;

    println!("Messages published successfully");
    Ok(())
}
```

--------------------------------

### ConnectionProperties

Source: https://context7.com/amqp-rs/lapin/llms.txt

Configuration options for AMQP connections, including locale, client properties, authentication, and recovery.

```APIDOC
## ConnectionProperties

### Description
Configuration options for AMQP connections including locale settings, client properties, authentication providers, and automatic recovery options. The builder pattern allows chaining multiple configuration options together.

### Method
N/A (Builder Pattern)

### Endpoint
N/A

### Parameters
#### Query Parameters
- **connection_name** (string) - Optional - Sets a descriptive connection name visible in RabbitMQ management.
- **client_property** (key: string, value: string) - Optional - Sets custom client properties.
- **locale** (string) - Optional - Configures the connection locale (e.g., `en_US`).
- **auto_recover** (boolean) - Optional - Enables automatic connection recovery. Defaults to false.
- **backoff_strategy** (ExponentialBuilder) - Optional - Configures a custom backoff strategy for reconnection attempts.

### Request Example
```rust
use lapin::{Connection, ConnectionProperties, Result};
use backon::ExponentialBuilder;

#[tokio::main]
async fn main() -> Result<()> {
    let properties = ConnectionProperties::default()
        .with_connection_name("order-processor".into())
        .with_client_property("application".into(), "order-service".into())
        .with_locale("en_US".into())
        .enable_auto_recover()
        .configure_backoff(|builder| {
            builder
                .with_max_times(10)
                .with_min_delay(std::time::Duration::from_millis(100))
                .with_max_delay(std::time::Duration::from_secs(30))
        });

    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", properties).await?;
    println!("Connected with auto-recovery enabled");
    Ok(())
}
```

### Response
#### Success Response (200)
- **properties** (ConnectionProperties) - The configured connection properties object.

#### Response Example
```rust
// Returns the configured ConnectionProperties object
```
```

--------------------------------

### Enable and Use AMQP Transactions

Source: https://context7.com/amqp-rs/lapin/llms.txt

Enables transaction mode, publishes messages within the transaction, and then either commits or rolls back the transaction. Ensure all publishes are within the transaction scope.

```rust
use lapin::{Connection, ConnectionProperties, BasicProperties, Result, options::*, types::FieldTable};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    channel.queue_declare("tx-queue".into(), QueueDeclareOptions::default(), FieldTable::default()).await?;

    // Enable transaction mode
    channel.tx_select().await?;
    println!("Transaction mode enabled");

    // Publish messages within transaction
    for i in 0..5 {
        channel
            .basic_publish(
                "".into(),
                "tx-queue".into(),
                BasicPublishOptions::default(),
                format!("Transaction message {}", i).as_bytes(),
                BasicProperties::default(),
            )
            .await?;
    }

    // Simulate decision to commit or rollback
    let should_commit = true;

    if should_commit {
        channel.tx_commit().await?;
        println!("Transaction committed - messages are now visible");
    } else {
        channel.tx_rollback().await?;
        println!("Transaction rolled back - messages discarded");
    }

    Ok(())
}
```

--------------------------------

### Channel::basic_consume

Source: https://context7.com/amqp-rs/lapin/llms.txt

Creates a consumer to receive messages from a queue. Requires manual acknowledgment of each message.

```APIDOC
## Channel::basic_consume

### Description
Creates a consumer that continuously receives messages from a queue. Returns a Consumer that implements the Stream trait, allowing iteration over incoming messages. Each message must be acknowledged, rejected, or negatively acknowledged.

### Method
`basic_consume`

### Endpoint
N/A (This is a library function, not an HTTP endpoint)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```rust
// Example usage within the lapin library context
let mut consumer = channel
    .basic_consume(
        "tasks".into(),           // Queue name
        "my-consumer".into(),     // Consumer tag (unique identifier)
        BasicConsumeOptions {
            no_local: false,
            no_ack: false,
            exclusive: false,
            nowait: false,
        },
        FieldTable::default(),
    )
    .await?;
```

### Response
#### Success Response (Stream of Deliveries)
- `Consumer` (Stream): Yields `Result<Delivery>` for each incoming message.

#### Response Example
```rust
// Processing messages from the consumer stream
while let Some(delivery_result) = consumer.next().await {
    match delivery_result {
        Ok(delivery) => {
            // Process delivery
            delivery.ack(BasicAckOptions::default()).await?;
        }
        Err(error) => {
            // Handle error
        }
    }
}
```
```

--------------------------------

### Enable Publisher Confirms and Publish Messages

Source: https://context7.com/amqp-rs/lapin/llms.txt

Enables publisher confirms mode for reliable publishing. Awaits individual or batched confirmations. Ensure the queue is declared before publishing.

```rust
use lapin::{
    Connection,
    ConnectionProperties,
    BasicProperties,
    Result,
    options::*,
    types::FieldTable,
};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    channel.queue_declare("reliable-queue".into(),
        QueueDeclareOptions { durable: true, ..Default::default() },
        FieldTable::default()).await?;

    // Enable publisher confirms
    channel.confirm_select(ConfirmSelectOptions::default()).await?;
    println!("Publisher confirms enabled");

    // Publish and wait for individual confirmation
    let confirm = channel
        .basic_publish(
            "".into(),
            "reliable-queue".into(),
            BasicPublishOptions::default(),
            b"Important message",
            BasicProperties::default().with_delivery_mode(2),
        )
        .await?   // Wait for message to be sent
        .await?;

    if confirm.is_ack() {
        println!("Message confirmed by server");
    } else {
        println!("Message was nacked!");
    }

    // Batch publishing: publish multiple, then wait for all confirms
    for i in 0..100 {
        channel
            .basic_publish(
                "".into(),
                "reliable-queue".into(),
                BasicPublishOptions::default(),
                format!("Batch message {}", i).as_bytes(),
                BasicProperties::default(),
            )
            .await?;
    }

    // Wait for all pending confirms at once
    let returned_messages = channel.wait_for_confirms().await?;
    if returned_messages.is_empty() {
        println!("All 100 messages confirmed");
    } else {
        println!("{} messages were returned", returned_messages.len());
    }

    Ok(())
}
```

--------------------------------

### Connection::create_channel

Source: https://context7.com/amqp-rs/lapin/llms.txt

Creates a new channel on an existing connection. Channels are lightweight and used for AMQP operations.

```APIDOC
## POST /create_channel

### Description
Creates a new channel on an existing connection. Channels are lightweight connections that share a single TCP connection. Most AMQP operations are performed on channels, and it's common to use separate channels for publishing and consuming.

### Method
POST

### Endpoint
/create_channel

### Parameters
#### Query Parameters
- **connection** (Connection) - Required - The existing AMQP connection object.

### Request Example
```rust
use lapin::{Connection, ConnectionProperties, Result};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect(
        "amqp://127.0.0.1:5672/%2f",
        ConnectionProperties::default(),
    )
    .await?;

    let publish_channel = conn.create_channel().await?;
    let consume_channel = conn.create_channel().await?;

    println!("Publisher channel ID: {}", publish_channel.id());
    println!("Consumer channel ID: {}", consume_channel.id());

    publish_channel.close(200, "Done publishing".into()).await?;
    consume_channel.close(200, "Done consuming".into()).await?;

    Ok(())
}
```

### Response
#### Success Response (200)
- **channel** (Channel) - A newly created AMQP channel object.

#### Response Example
```rust
// Successful creation returns a Channel object
```
```

--------------------------------

### Set QoS Prefetch Count

Source: https://context7.com/amqp-rs/lapin/llms.txt

Configure the prefetch count for a channel to limit unacknowledged messages delivered to a consumer. Set this before creating a consumer to prevent overwhelming it.

```rust
use futures_lite::StreamExt;
use lapin::{Connection, ConnectionProperties, Result, options::*,
 types::FieldTable};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    // Set prefetch count before creating consumer
    // Only deliver 10 unacknowledged messages at a time
    channel
        .basic_qos(
            10,  // Prefetch count
            BasicQosOptions {
                global: false,  // Per-consumer (not per-channel)
            },
        )
        .await?;

    channel.queue_declare("tasks".into(), QueueDeclareOptions::default(), FieldTable::default()).await?;

    let mut consumer = channel
        .basic_consume(
            "tasks".into(),
            "worker".into(),
            BasicConsumeOptions::default(),
            FieldTable::default(),
        )
        .await?;

    println!("Consumer with prefetch=10 started");

    while let Some(delivery) = consumer.next().await {
        if let Ok(delivery) = delivery {
            // Process message (prefetch ensures we don't get overwhelmed)
            println!("Processing delivery {}", delivery.delivery_tag);
            delivery.ack(BasicAckOptions::default()).await?;
        }
    }

    Ok(())
}
```

--------------------------------

### Channel::basic_publish

Source: https://context7.com/amqp-rs/lapin/llms.txt

Publishes a message to an exchange with a routing key. The message consists of a payload and optional properties (headers, content type, delivery mode, etc.). Returns a PublisherConfirm that can be awaited when publisher confirms are enabled.

```APIDOC
## Channel::basic_publish

### Description
Publishes a message to an exchange with a routing key. The message consists of a payload and optional properties (headers, content type, delivery mode, etc.). Returns a PublisherConfirm that can be awaited when publisher confirms are enabled.

### Method
POST (or equivalent AMQP command)

### Endpoint
N/A (This is a library function, not a direct HTTP endpoint)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None (Parameters are passed as function arguments)

### Request Example
```rust
use lapin::{Connection, ConnectionProperties, BasicProperties, Result, options::*, types::FieldTable};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    // Ensure queue exists
    channel.queue_declare("tasks".into(), QueueDeclareOptions::default(), FieldTable::default()).await?;

    // Simple publish to default exchange (routing key = queue name)
    let payload = b"Hello, World!";
    channel
        .basic_publish(
            "".into(),           // Default exchange
            "tasks".into(),      // Routing key (queue name for default exchange)
            BasicPublishOptions::default(),
            payload,
            BasicProperties::default(),
        )
        .await?;

    // Publish with message properties
    let order_data = r#"{"order_id": 12345, "amount": 99.99}"#;
    channel
        .basic_publish(
            "".into(),
            "tasks".into(),
            BasicPublishOptions::default(),
            order_data.as_bytes(),
            BasicProperties::default()
                .with_content_type("application/json".into())
                .with_delivery_mode(2)  // Persistent
                .with_priority(5)
                .with_message_id("msg-12345".into())
                .with_correlation_id("req-67890".into()),
        )
        .await?;

    // Publish with mandatory flag (returns message if unroutable)
    channel
        .basic_publish(
            "".into(),
            "tasks".into(),
            BasicPublishOptions {
                mandatory: true,  // Return message if no queue is bound
                immediate: false,
            },
            payload,
            BasicProperties::default(),
        )
        .await?;

    println!("Messages published successfully");
    Ok(())
}
```

### Response
#### Success Response (200 OK or equivalent AMQP status)
Indicates the publish operation was successful. If publisher confirms are enabled, a `PublisherConfirm` object is returned which can be awaited.

#### Response Example
None (The operation is confirmed by the `await` completing without error, or by awaiting the `PublisherConfirm`)
```

--------------------------------

### Bind Queue to Exchange with Routing Keys

Source: https://context7.com/amqp-rs/lapin/llms.txt

Binds a queue to an exchange using different routing key patterns including exact matches, single-word wildcards (*), and multi-word wildcards (#). Ensure the exchange and queue are declared before binding.

```rust
use lapin::{Connection, ConnectionProperties, ExchangeKind, Result, options::*, types::FieldTable};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    // Setup exchange and queue
    channel.exchange_declare("events".into(), ExchangeKind::Topic,
        ExchangeDeclareOptions { durable: true, ..Default::default() },
        FieldTable::default()).await?;

    channel.queue_declare("order-events".into(),
        QueueDeclareOptions { durable: true, ..Default::default() },
        FieldTable::default()).await?;

    // Bind with exact routing key
    channel
        .queue_bind(
            "order-events".into(),
            "events".into(),
            "order.created".into(),
            QueueBindOptions::default(),
            FieldTable::default(),
        )
        .await?;

    // Bind with wildcard pattern (* matches one word)
    channel
        .queue_bind(
            "order-events".into(),
            "events".into(),
            "order.*".into(),
            QueueBindOptions::default(),
            FieldTable::default(),
        )
        .await?;

    // Bind with multi-level wildcard (# matches zero or more words)
    channel
        .queue_bind(
            "order-events".into(),
            "events".into(),
            "order.#".into(),
            QueueBindOptions::default(),
            FieldTable::default(),
        )
        .await?;

    println!("Queue bound to exchange with multiple routing patterns");
    Ok(())
}
```

--------------------------------

### Enable Automatic Connection Recovery

Source: https://github.com/amqp-rs/lapin/blob/main/README.md

Enable automatic connection recovery by calling `enable_auto_recover()` on `ConnectionProperties`. This is useful for handling network failures.

```rust
let properties = ConnectionProperties::default().enable_auto_recover(); // you might also want to configure the backoff for the TCP connection itself
// connect using properties.
```

--------------------------------

### Channel::queue_declare

Source: https://context7.com/amqp-rs/lapin/llms.txt

Declares a queue on the server. Options control durability, exclusivity, and auto-deletion behavior. Returns a Queue object.

```APIDOC
## Channel::queue_declare

### Description
Declares a queue on the server. If the queue already exists with identical parameters, this is a no-op. Returns a Queue object containing the queue name, message count, and consumer count. Options control durability, exclusivity, and auto-deletion behavior.

### Method
POST (Implicit, as it's an operation on the channel)

### Endpoint
/amqp/queues

### Parameters
#### Query Parameters
- **name** (string) - Required - The name of the queue. An empty string will result in a server-generated unique name.
- **options** (object) - Optional - Configuration for the queue:
  - **durable** (boolean) - Optional - If true, the queue will survive broker restarts. Defaults to false.
  - **exclusive** (boolean) - Optional - If true, the queue will only be accessible by the current connection. Defaults to false.
  - **auto_delete** (boolean) - Optional - If true, the queue will be deleted when the last consumer disconnects. Defaults to false.
  - **passive** (boolean) - Optional - If true, the server will only check if the queue exists and return an error if it doesn't. Defaults to false.
  - **nowait** (boolean) - Optional - If true, the server will not respond to the request. Defaults to false.

#### Request Body
Not applicable for this operation, parameters are passed as query parameters or within the method signature.

### Request Example
```rust
channel
    .queue_declare(
        "tasks".into(),
        QueueDeclareOptions {
            durable: true,
            exclusive: false,
            auto_delete: false,
            passive: false,
            nowait: false,
        },
        FieldTable::default(),
    )
    .await?;
```

### Response
#### Success Response (200)
- **queue_name** (string) - The name of the declared queue.
- **message_count** (integer) - The number of messages currently in the queue.
- **consumer_count** (integer) - The number of active consumers on the queue.

#### Response Example
```json
{
  "queue_name": "tasks",
  "message_count": 0,
  "consumer_count": 0
}
```
```

--------------------------------

### Enable Parallel Message Processing with Consumer::set_delegate

Source: https://context7.com/amqp-rs/lapin/llms.txt

Sets a delegate handler that is automatically spawned for each incoming message, enabling parallel processing. Useful for high-throughput scenarios where messages can be processed independently. Each delivery is handled in its own async task.

```rust
use lapin::
    Connection, ConnectionProperties, Result,
    message::DeliveryResult,
    options::*, types::FieldTable,
};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    channel.queue_declare("tasks".into(), QueueDeclareOptions::default(), FieldTable::default()).await?;

    // Create consumer and set delegate for parallel processing
    let consumer = channel
        .basic_consume(
            "tasks".into(),
            "parallel-consumer".into(),
            BasicConsumeOptions::default(),
            FieldTable::default(),
        )
        .await?;

    // Each message is processed in a separate spawned task
    consumer.set_delegate(move |delivery: DeliveryResult| async move {
        match delivery {
            Ok(Some(delivery)) => {
                let data = String::from_utf8_lossy(&delivery.data);
                println!("Processing: {}", data);

                // Simulate work
                tokio::time::sleep(std::time::Duration::from_millis(100)).await;

                // Acknowledge after processing
                if let Err(e) = delivery.ack(BasicAckOptions::default()).await {
                    eprintln!("Failed to ack: {:?}", e);
                }
            }
            Ok(None) => println!("Consumer cancelled"),
            Err(error) => eprintln!("Consumer error: {:?}", error),
        }
    });

    println!("Delegate set, messages will be processed in parallel");

    // Keep the application running
    tokio::time::sleep(std::time::Duration::from_secs(60)).await;
    Ok(())
}
```

--------------------------------

### Declare AMQP Queues in Rust

Source: https://context7.com/amqp-rs/lapin/llms.txt

Declares queues with specified options like durability, exclusivity, and auto-deletion. Use an empty string for the queue name to have the server generate a unique name.

```rust
use lapin::{Connection, ConnectionProperties, Result, options::*, types::FieldTable};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    // Declare a durable queue that survives broker restarts
    let queue = channel
        .queue_declare(
            "tasks".into(),
            QueueDeclareOptions {
                durable: true,      // Queue survives broker restart
                exclusive: false,   // Other connections can access
                auto_delete: false, // Queue persists when consumers disconnect
                passive: false,     // Create if doesn't exist
                nowait: false,      // Wait for server confirmation
            },
            FieldTable::default(),
        )
        .await?;

    println!("Queue '{}' declared", queue.name());
    println!("Messages in queue: {}", queue.message_count());
    println!("Active consumers: {}", queue.consumer_count());

    // Declare a temporary exclusive queue (auto-named by server)
    let temp_queue = channel
        .queue_declare(
            "".into(), // Empty name = server generates unique name
            QueueDeclareOptions {
                exclusive: true,    // Only this connection can access
                auto_delete: true,  // Delete when connection closes
                ..Default::default()
            },
            FieldTable::default(),
        )
        .await?;

    println!("Temporary queue: {}", temp_queue.name());
    Ok(())
}
```

--------------------------------

### Channel::basic_qos

Source: https://context7.com/amqp-rs/lapin/llms.txt

Sets the Quality of Service (QoS) prefetch count for a channel. This limits the number of unacknowledged messages the server will deliver to a consumer, helping to manage load and prevent consumers from being overwhelmed.

```APIDOC
## Channel::basic_qos

### Description
Sets the Quality of Service (QoS) prefetch count, limiting how many unacknowledged messages the server will deliver to a consumer. This is essential for load balancing and preventing consumers from being overwhelmed.

### Method
`Channel::basic_qos`

### Parameters
#### Query Parameters
- **prefetch_count** (integer) - Required - The number of unacknowledged messages to prefetch.
- **options** (BasicQosOptions) - Required - Options for QoS settings.
  - **global** (boolean) - Optional - If true, applies QoS settings globally to the connection; otherwise, applies per-consumer.

### Request Example
```rust
channel
    .basic_qos(
        10,  // Prefetch count
        lapin::options::BasicQosOptions {
            global: false,  // Per-consumer (not per-channel)
        },
    )
    .await?;
```

### Response
#### Success Response (200)
Indicates that the QoS settings were successfully applied.

#### Response Example
(No specific response body is detailed, success is indicated by the absence of an error.)
```

--------------------------------

### Handle Message Acknowledgments

Source: https://context7.com/amqp-rs/lapin/llms.txt

Implement message acknowledgment logic using ack, nack, and reject. Acknowledge successful processing, nack with requeue for temporary failures, and reject for permanent failures.

```rust
use futures_lite::StreamExt;
use lapin::{Connection, ConnectionProperties, Result, options::*,
 types::FieldTable};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    channel.queue_declare("tasks".into(), QueueDeclareOptions::default(), FieldTable::default()).await?;

    let mut consumer = channel
        .basic_consume("tasks".into(), "worker".into(), BasicConsumeOptions::default(), FieldTable::default())
        .await?;

    while let Some(delivery) = consumer.next().await {
        if let Ok(delivery) = delivery {
            let data = String::from_utf8_lossy(&delivery.data);

            // Simulated processing result
            let processing_result: Result<(), &str> = if data.contains("error") {
                Err("Processing failed")
            } else {
                Ok(())
            };

            match processing_result {
                Ok(())
                    => {
                    // Success: acknowledge the message
                    delivery.ack(BasicAckOptions::default()).await?;
                    println!("Message acknowledged");
                }
                Err(e) if data.contains("retry")
                    => {
                    // Temporary failure: negative ack with requeue
                    delivery
                        .nack(BasicNackOptions {
                            multiple: false,  // Only this message
                            requeue: true,    // Put back in queue
                        })
                        .await?;
                    println!("Message requeued for retry: {}", e);
                }
                Err(e) => {
                    // Permanent failure: reject without requeue
                    delivery
                        .reject(BasicRejectOptions {
                            requeue: false,  // Don't requeue (goes to DLX if configured)
                        })
                        .await?;
                    println!("Message rejected: {}", e);
                }
            }
        }
    }

    Ok(())
}
```

--------------------------------

### Enable Automatic Connection Recovery in Lapin

Source: https://context7.com/amqp-rs/lapin/llms.txt

Enables automatic reconnection and topology recovery after connection failures. This snippet demonstrates how to configure connection properties to enable auto-recovery and retry mechanisms.

```rust
use async_rs::{Runtime, traits::*};
use futures_lite::StreamExt;
use lapin::{Connection, ConnectionProperties, Result, options::*,
 types::FieldTable};

#[tokio::main]
async fn main() -> Result<()> {
    let runtime = Runtime::tokio_current();

    // Enable auto-recovery in connection properties
    let conn = Connection::connect_with_runtime(
        "amqp://127.0.0.1:5672/%2f",
        ConnectionProperties::default()
            .enable_auto_recover()  // Enable automatic recovery
            .configure_backoff(|b| b.with_max_times(100)),  // Retry up to 100 times
        runtime.clone(),
    )
    .await?;

    let channel = conn.create_channel().await?;
    channel.queue_declare("resilient-queue".into(),
        QueueDeclareOptions { durable: true, ..Default::default() },
        FieldTable::default()).await?;

    let mut consumer = channel
        .basic_consume("resilient-queue".into(), "resilient-consumer".into(),
            BasicConsumeOptions::default(), FieldTable::default())
        .await?;

    // Consumer loop with recovery handling
    loop {
        match consumer.next().await {
            Some(Ok(delivery)) => {
                println!("Received: {}", String::from_utf8_lossy(&delivery.data));
                delivery.ack(BasicAckOptions::default()).await?;
            }
            Some(Err(error)) => {
                // Wait for recovery to complete
                println!("Error occurred, waiting for recovery...");
                channel.wait_for_recovery(error).await?;
                println!("Recovery complete, resuming...");
            }
            None => {
                println!("Consumer cancelled");
                break;
            }
        }
    }

    Ok(())
}
```

--------------------------------

### Synchronously Retrieve a Single Message with basic_get

Source: https://context7.com/amqp-rs/lapin/llms.txt

Retrieves a single message from a queue synchronously using a polling approach. The queue must be declared first. Acknowledgment is required if `no_ack` is false.

```rust
use lapin::{Connection, ConnectionProperties, Result, options::*, types::FieldTable};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    channel.queue_declare("tasks".into(), QueueDeclareOptions::default(), FieldTable::default()).await?;

    // Poll for a single message
    match channel.basic_get("tasks".into(), BasicGetOptions { no_ack: false }).await? {
        Some(message) => {
            println!("Got message: {}", String::from_utf8_lossy(&message.data));
            println!("Messages remaining in queue: {}", message.message_count);

            // Must acknowledge if no_ack is false
            message.ack(BasicAckOptions::default()).await?;
        }
        None => {
            println!("Queue is empty");
        }
    }

    // Polling loop example
    loop {
        if let Some(msg) = channel.basic_get("tasks".into(), BasicGetOptions { no_ack: false }).await? {
            println!("Processing: {}", String::from_utf8_lossy(&msg.data));
            msg.ack(BasicAckOptions::default()).await?;
        } else {
            // No messages, wait before polling again
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        }
    }
}
```

--------------------------------

### Declare AMQP Exchanges in Rust

Source: https://context7.com/amqp-rs/lapin/llms.txt

Declares exchanges of different types (Direct, Fanout, Topic) with durable options. Exchanges are used to route messages to queues.

```rust
use lapin::{Connection, ConnectionProperties, ExchangeKind, Result, options::*, types::FieldTable};

#[tokio::main]
async fn main() -> Result<()> {
    let conn = Connection::connect("amqp://127.0.0.1:5672/%2f", ConnectionProperties::default()).await?;
    let channel = conn.create_channel().await?;

    // Declare a direct exchange for point-to-point messaging
    channel
        .exchange_declare(
            "orders".into(),
            ExchangeKind::Direct,
            ExchangeDeclareOptions {
                durable: true,
                ..Default::default()
            },
            FieldTable::default(),
        )
        .await?;

    // Declare a fanout exchange for broadcasting
    channel
        .exchange_declare(
            "notifications".into(),
            ExchangeKind::Fanout,
            ExchangeDeclareOptions {
                durable: true,
                ..Default::default()
            },
            FieldTable::default(),
        )
        .await?;

    // Declare a topic exchange for pattern-based routing
    channel
        .exchange_declare(
            "events".into(),
            ExchangeKind::Topic,
            ExchangeDeclareOptions {
                durable: true,
                ..Default::default()
            },
            FieldTable::default(),
        )
        .await?;

    println!("Exchanges declared successfully");
    Ok(())
}
```