Lapin

https://docs.rs/lapin/latest
Admin
Lapin is a Rust AMQP 0.9.1 client library for RabbitMQ that provides async/await support for message...
Tokens:169,187
Snippets:2,881
Trust Score:9.7
Update:4 months ago
Show doc for...
Tokens:
Raw
### Quick start

Source: https://docs.rs/lapin/latest/src/lapin/lib.rs.html

Example of connecting to RabbitMQ, declaring a queue, publishing a message, and consuming it.

```rust
#![deny(missing_docs, missing_debug_implementations, unsafe_code)]
#![warn(unreachable_pub, unused_qualifications, unused_lifetimes)]
#![warn(
    clippy::must_use_candidate,
    clippy::unwrap_in_result,
    clippy::panic_in_result_fn
)]

//! An async AMQP 0-9-1 client library targeting RabbitMQ.
//!
//! # Core concepts
//!
//! **[`Connection`]** — a single TCP socket to the broker. One process
//! typically creates one connection and reuses it throughout its lifetime.
//!
//! **[`Channel`]** — a lightweight virtual connection multiplexed over a
//! [`Connection`]. All AMQP operations (declaring queues, publishing, consuming, …) are performed through channels. Open as many as you need; they are cheap.
//!
//! **[`Consumer`]** — an async `Stream` of [`message::Delivery`] values obtained by calling [`Channel::basic_consume`]. Each delivery must be explicitly acknowledged once processed.
//!
//! **[`PublisherConfirm`]** — a future returned by [`Channel::basic_publish`] that resolves to a [`Confirmation`] once the broker has acknowledged the message (requires [`Channel::confirm_select`]).
//!
//! # Quick start
//!
//! ```rust,no_run
//! use async_rs::traits::Executor;
//! use futures_lite::stream::StreamExt;
//! use lapin::
//!     options::*,
//!     types::FieldTable,
//!     BasicProperties,
//!     Connection,
//!     ConnectionProperties,
//!     Result,
//! };
//!
//! fn main() -> Result<()> {
//!     let addr = std::env::var("AMQP_ADDR")
//!         .unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
//!     let runtime = lapin::runtime::default_runtime()?;
//!
//!     runtime.clone().block_on(async move {
//!         let conn = Connection::connect(&addr, ConnectionProperties::default()).await?;
//!
//!         let channel = conn.create_channel().await?;
//!
//!         channel
//!             .queue_declare("hello".into(), QueueDeclareOptions::durable(), FieldTable::default())
//!             .await?;
//!
//!         channel
//!             .basic_publish(
//!                 "".into(),
//!                 "hello".into(),
//!                 BasicPublishOptions::default(),
//!                 b"Hello, world!",
//!                 BasicProperties::default(),
//!             )
//!             .await?
//!             .await?;
//!
//!         let mut consumer = channel
//!             .basic_consume(
//!                 "hello".into(),
//!                 "my_consumer".into(),
//!                 BasicConsumeOptions::default(),
//!                 FieldTable::default(),
//!             )
//!             .await?;
//!
//!         while let Some(delivery) = consumer.next().await {
//!             let delivery = delivery?;
//!             delivery.ack(BasicAckOptions::default()).await?;
//!         }
//!         Ok(())
//!     })
//! }
//! ```
//!
//! # Automatic connection recovery
//!
//! Enable recovery in [`ConnectionProperties`] to automatically reconnect and
//! replay topology (exchanges, queues, bindings, consumers) after a network
//! failure:
//!
//! ```rust,no_run
//! use lapin::ConnectionProperties;
//!
//! let props = ConnectionProperties::default().enable_auto_recover();
//! // then pass `props` to Connection::connect(…)
//! ```
//!
//! After catching an error from a channel operation, call
//! [`Channel::wait_for_recovery`] to block until the connection has been
//! re-established:
//!
//! ```rust,no_run
//! # use lapin::{Channel, Error, Result};
//! # async fn example(channel: Channel, error: Error) -> Result<()> {
//! channel.wait_for_recovery(error).await?;
//! # Ok(())
//! # }
//! ```
//!
//! # Feature flags
//!
//! ## Async runtime (pick exactly one)
//!
//! | Flag | Notes |
//! |------|-------|
//! | `tokio` *(default)* | Requires a Tokio runtime |
//! | `smol` | Uses the smol executor |
//! | `async-global-executor` | Uses async-global-executor |
//!
//! ## TLS backend (pick at most one; `rustls` is the default)
//!
//! | Flag | Notes |
//! |------|-------|
//! | `rustls` *(default)* | TLS via rustls |
//! | `native-tls` | TLS via the platform's native library |
//! | `openssl` | TLS via OpenSSL |
//!
//! ## Rustls certificate store (only when `rustls` is active)
//!
//! | Flag | Notes |
//! |------|-------|
//! | `rustls-platform-verifier` *(default)* | Uses the platform trust store |
//! | `rustls-native-certs` | Loads native root certificates |
//! | `rustls-webpki-roots-certs` | Uses the webpki bundled root set |
//!
//! ## Rustls crypto provider (at least one must be enabled)
//!
//! | Flag | Notes |
//! |------|-------|
//! | `rustls--aws_lc_rs` *(default)* | Uses aws-lc-rs |
//! | `rustls--ring` | Uses ring (more portable) |

```

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

### Quick start

Source: https://docs.rs/lapin/latest/index.html

A basic example demonstrating how to connect to a RabbitMQ broker, declare a queue, publish a message, and consume messages from the queue.

```rust
use async_rs::traits::Executor;
use futures_lite::stream::StreamExt;
use lapin::
    options::*,
    types::FieldTable,
    BasicProperties,
    Connection,
    ConnectionProperties,
    Result,
;

fn main() -> Result<()> {
    let addr = std::env::var("AMQP_ADDR")
        .unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
    let runtime = lapin::runtime::default_runtime()?;

    runtime.clone().block_on(async move {
        let conn = Connection::connect(&addr, ConnectionProperties::default()).await?;

        let channel = conn.create_channel().await?;

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

        channel
            .basic_publish(
                "".into(),
                "hello".into(),
                BasicPublishOptions::default(),
                b"Hello, world!",
                BasicProperties::default(),
            )
            .await?;

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

        while let Some(delivery) = consumer.next().await {
            let delivery = delivery?;
            delivery.ack(BasicAckOptions::default()).await?;
        }
        Ok(())
    })
}
```

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

### IoLoop::finish_setup method

Source: https://docs.rs/lapin/latest/src/lapin/io_loop.rs.html

Completes the setup process for the I/O loop once a connection is established. It adjusts buffer sizes based on the negotiated frame max, starts the heartbeat if configured, and updates the status to Connected.

```rust
fn finish_setup(&mut self) -> Result<bool> {
    if self.connection_status.connected() {
        let frame_max = self.configuration.frame_max();
        self.frame_size = std::cmp::max(self.frame_size, frame_max);
        self.receive_buffer
            .grow(FRAMES_STORAGE * self.frame_size as usize);
        self.send_buffer
            .grow(FRAMES_STORAGE * self.frame_size as usize);
        let heartbeat = self.configuration.heartbeat();
        if heartbeat != 0 {
            let heartbeat = Duration::from_millis(u64::from(heartbeat) * 500); // * 1000 (ms) / 2 (half the negotiated timeout)
            self.internal_rpc.start_heartbeat(heartbeat);
        }
        self.status = Status::Connected;
        self.global_backoff = self.backoff.build();
        self.first_connection = false;
    }
    Ok(true)
}
```

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

### Connection Initialization and Start

Source: https://docs.rs/lapin/latest/src/lapin/connection.rs.html

This snippet shows the initialization of various components like InternalRPC, Events, and Channels, followed by the setup and start of the IoLoop and the Connection itself.

```rust
let internal_rpc = InternalRPC::new(
            runtime.clone(),
            heartbeat.clone(),
            secret_update,
            frames.clone(),
            socket_state.handle(),
        );
        let events = Events::new();
        let channels = Channels::new(
            configuration.clone(),
            status.clone(),
            socket_state.handle(),
            internal_rpc.handle(),
            frames.clone(),
            events.clone(),
        );
        let channel0 = channels.channel0();
        let conn = Connection::new(configuration, status, internal_rpc.handle(), events);
        let io_loop = IoLoop::new(
            conn.status.clone(),
            conn.configuration.negotiated_config.clone(),
            channels.clone(),
            internal_rpc.handle(),
            frames,
            socket_state,
            heartbeat,
            runtime,
            connect,
            uri,
            conn.configuration().backoff,
        );

        internal_rpc.start(channels);
        conn.io_loop.register(io_loop.start()?);
        conn.start(channel0).await
```

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

### Consumer Example

Source: https://docs.rs/lapin/latest/src/lapin/consumer.rs.html

Example of how to set up a consumer and process messages. This example connects to an AMQP server, sets up a consumer with default options, and then iterates over incoming messages.

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

fn main() -> Result<()> {
    let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
    let runtime = lapin::runtime::default_runtime().unwrap();

    runtime.clone().block_on(async move {
        let conn = Connection::connect_with_runtime(
            &addr,
            ConnectionProperties::default(),
            runtime,
        )
        .await?;

```

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

### Basic Consumer Example

Source: https://docs.rs/lapin/latest/src/lapin/consumer.rs.html

An example of how to set up a basic consumer to receive messages.

```rust
let channel = conn.create_channel().await?;
let mut consumer = channel
    .basic_consume(
        "hello".into(),
        "my_consumer".into(),
        BasicConsumeOptions::default(),
        FieldTable::default(),
    )
    .await?;

while let Some(delivery) = consumer.next().await {
    let delivery = delivery.expect("error in consumer");
    delivery
        .ack(BasicAckOptions::default())
        .await?;
}
Ok(())
}
```

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

### TokenAuthProvider Example

Source: https://docs.rs/lapin/latest/src/lapin/auth.rs.html

Example of creating a TokenAuthProvider with a DefaultTokenProvider that never expires.

```rust
let auth_provider: TokenAuthProvider<_> = DefaultTokenProvider::new(
    |_token| None, // never expire
    || Ok("my new valid token".into()),
).into();
```

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

### ConnectionBuilder Example

Source: https://docs.rs/lapin/latest/src/lapin/connection_builder.rs.html

Demonstrates how to create a connection using DefaultConnectionBuilder, setting the URI and connecting.

```rust
use lapin::{DefaultConnectionBuilder, Result};
# async fn example() -> Result<()> {
let conn = DefaultConnectionBuilder::new()?
    .with_uri_str("amqp://localhost".into())
    .connect()
    .await?;
# Ok(())
# }
```

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

### Example Usage

Source: https://docs.rs/lapin/latest/lapin/auth/struct.DefaultTokenProvider.html

Example demonstrating how to create a DefaultTokenProvider and convert it into a TokenAuthProvider.

```rust
use lapin::auth::{DefaultTokenProvider, TokenAuthProvider};

let auth_provider: TokenAuthProvider<_> = DefaultTokenProvider::new(
    |_token| None, // never expire
    || Ok("my new valid token".into()),
).into();
```

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

### Connection Start OK Implementation

Source: https://docs.rs/lapin/latest/src/lapin/opt/rustwide/target/x86_64-unknown-linux-gnu/debug/build/lapin-0aa65c3f4b556b6e/out/channel.rs.html

Rust code for sending the connection.start-ok response.

```Rust
async fn connection_start_ok(&self, client_properties: FieldTable, mechanism: ShortString, response: LongString, locale: ShortString, conn_resolver: PromiseResolver<Connection>, connection: Connection, auth_provider: Arc<dyn AuthProvider>) -> Result<()> {
let (promise, resolver) = Promise::new("connection.start-ok");
let method = AMQPClass::Connection(protocol::connection::AMQPMethod::StartOk (
protocol::connection::StartOk {
client_properties,
mechanism,
response,
locale,
}
));

self.send_method_frame(method, Box::new(resolver.clone()), Some(ExpectedReply(Reply::ConnectionStep(ConnectionStep::StartOk(conn_resolver, connection, auth_provider)), Box::new(resolver))), None);
promise.await
```

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

### Clone Implementation

Source: https://docs.rs/lapin/latest/lapin/protocol/connection/struct.Start.html

Implementation of the Clone trait for the Start struct.

```rust
fn clone(&self) -> Start
```

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

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

### PartialEq Implementation

Source: https://docs.rs/lapin/latest/lapin/protocol/connection/struct.Start.html

Implementation of the PartialEq trait for the Start struct.

```rust
fn eq(&self, other: &Start) -> bool
```

```rust
fn ne(&self, other: &Rhs) -> bool
```

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

### Start Struct Definition

Source: https://docs.rs/lapin/latest/lapin/protocol/connection/struct.Start.html

The Rust struct definition for Start, including its fields.

```rust
pub struct Start {
    pub version_major: u8,
    pub version_minor: u8,
    pub server_properties: FieldTable,
    pub mechanisms: LongString,
    pub locales: LongString,
}
```

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

### gen_start

Source: https://docs.rs/lapin/latest/lapin/protocol/connection/fn.gen_start.html

Serialize start (Generated)

```rust
pub fn gen_start<'a, W>(method: &'a Start) -> impl SerializeFn<W> + 'a
where
    W: Write + BackToTheBuffer + 'a,

```

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

### Default Implementation

Source: https://docs.rs/lapin/latest/lapin/protocol/connection/struct.Start.html

Implementation of the Default trait for the Start struct.

```rust
fn default() -> Start
```

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

### Receive Connection Start

Source: https://docs.rs/lapin/latest/src/lapin/opt/rustwide/target/x86_64-unknown-linux-gnu/debug/build/lapin-0aa65c3f4b556b6e/out/channel.rs.html

Rust code for handling the reception of a connection.start method.

```Rust
fn receive_connection_start(&self, method: protocol::connection::Start) -> Result<()> {
self.assert_channel0(
method.get_amqp_class_id(),
method.get_amqp_method_id(),
)?;
if !self.status.can_receive_messages() {
return Err(self.status.state_error("connection.start"));
}

match self.frames.find_connection_step(self.id) {
Some(step) => self.on_connection_start_received(method, step),
None => self.connection_process_error(self.connection_status.state(), None, None),
}
}
```

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

### into_ok() - Example

Source: https://docs.rs/lapin/latest/lapin/type.Result.html

Illustrates the experimental into_ok() method, which returns the Ok value without panicking.

```rust
fn only_good_news() -> Result<String, !> {
    Ok("this is fine".into())
}

let s: String = only_good_news().into_ok();
println!("{s}");
```

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

### expect

Source: https://docs.rs/lapin/latest/lapin/type.Result.html

Example of using expect to retrieve the Ok value, panicking if Err.

```rust
let x: Result<u32, &str> = Err("emergency failure");
x.expect("Testing expect"); // panics with `Testing expect: emergency failure`
```

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

### Connection Start OK Logic

Source: https://docs.rs/lapin/latest/src/lapin/channel.rs.html

Handles the client properties and capabilities when initiating a connection, including product, version, platform, and specific AMQP capabilities.

```rust
if !configuration.amqp_client_properties.contains_key("product")
                    || !configuration.amqp_client_properties.contains_key("version")
                {
                    configuration.amqp_client_properties.insert(
                        "product".into(),
                        AMQPValue::LongString(env!("CARGO_PKG_NAME").into()),
                    );
                    configuration.amqp_client_properties.insert(
                        "version".into(),
                        AMQPValue::LongString(env!("CARGO_PKG_VERSION").into()),
                    );
                }

                configuration
                    .amqp_client_properties
                    .insert("platform".into(), AMQPValue::LongString("rust".into()));

                let mut capabilities = FieldTable::default();
                capabilities.insert("publisher_confirms".into(), true.into());
                capabilities.insert("exchange_exchange_bindings".into(), true.into());
                capabilities.insert("basic.nack".into(), true.into());
                capabilities.insert("consumer_cancel_notify".into(), true.into());
                capabilities.insert("connection.blocked".into(), true.into());
                capabilities.insert("consumer_priorities".into(), true.into());
                capabilities.insert("authentication_failure_close".into(), true.into());
                capabilities.insert("per_consumer_qos".into(), true.into());
                capabilities.insert("direct_reply_to".into(), true.into());

                configuration
                    .amqp_client_properties
                    .insert("capabilities".into(), AMQPValue::FieldTable(capabilities));

                let auth_starter = auth_provider
                    .auth_starter()
                    .map_err(ErrorKind::AuthProviderError)?;
                let channel = self.clone();
                let client_properties = configuration.amqp_client_properties.clone();
                self.internal_rpc.spawn(async move {
                    channel
                        .connection_start_ok(
                            client_properties,
                            mechanism,
                            auth_starter,
                            locale,
                            resolver,
                            connection,
                            auth_provider,
                        )
                        .await
                });
                Ok(())

```

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

### Connection Start Method

Source: https://docs.rs/lapin/latest/src/lapin/connection.rs.html

This method initiates the connection by setting the status to connecting, sending the protocol header, and waiting for a promise to be resolved.

```rust
pub(crate) async fn start(self, channel0: Channel) -> Result<Self> {
        let (promise, resolver) = Promise::new("ProtocolHeader");

        trace!("Set connection as connecting");
        self.status.clone().set_connecting()?;

        trace!("Sending protocol header to server");
        channel0.send_frame(
            AMQPFrame::ProtocolHeader(ProtocolVersion::amqp_0_9_1()),
            Box::new(resolver.clone()),
            Some(ExpectedReply(
                Reply::ConnectionStep(ConnectionStep::ProtocolHeader(resolver.clone(), self)),
                Box::new(resolver),
            )),
            None,
        );

        trace!("Sent protocol header to server, waiting for connection flow");
        promise.await
    }
```

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

### Debug Implementation

Source: https://docs.rs/lapin/latest/lapin/protocol/connection/struct.Start.html

Implementation of the Debug trait for the Start struct.

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

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

### Example of creating and using ConnectionProperties

Source: https://docs.rs/lapin/latest/lapin/struct.ConnectionProperties.html

Demonstrates how to create default ConnectionProperties, customize them using builder methods, and establish a connection.

```rust
use lapin::{Connection, ConnectionProperties, Result};
let props = ConnectionProperties::default()
    .with_connection_name("my-app".into())
    .enable_auto_recover();
let conn = Connection::connect("amqp://localhost", props).await?;
```

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

### CloneToUninit Trait Implementation

Source: https://docs.rs/lapin/latest/lapin/protocol/connection/struct.Start.html

Blanket implementation of the CloneToUninit trait.

```rust
unsafe fn clone_to_uninit(&self, dest: *mut u8)
```

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

### Starting Connection Recovery

Source: https://docs.rs/lapin/latest/src/lapin/channels.rs.html

This snippet demonstrates the asynchronous process of starting connection recovery, including resetting killswitches, dropping frames, and initiating reconnection for the connection and its channels.

```rust
pub(crate) async fn start_recovery(&self) -> Result<()> {
        let channels = self.read().channels.values().cloned().collect::<Vec<_>>();

        self.connection_killswitch.reset();
        self.frames.drop_poison();
        self.channel0.update_recovery();
        self.channel0.finalize_connection();

        Connection::for_reconnect(
            self.configuration.clone(),
            self.connection_status.clone(),
            self.internal_rpc.clone(),
            self.events.clone(),
        )
        .start(self.channel0())
        .await?;

        trace!("Connection recovered, now recovering channels");

        // TODO: use future::join! when stable
    }
```

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

### Example Usage

Source: https://docs.rs/lapin/latest/lapin/struct.ConnectionBuilder.html

Demonstrates how to create a default connection builder, set the URI, and establish an asynchronous connection to an AMQP broker.

```rust
use lapin::{DefaultConnectionBuilder, Result};
let conn = DefaultConnectionBuilder::new()? 
    .with_uri_str("amqp://localhost".into())
    .connect()
    .await?;
```

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

### ConnectionProperties Example

Source: https://docs.rs/lapin/latest/src/lapin/connection_properties.rs.html

Build ConnectionProperties with Default::default and then chain the with_* / enable_* builder methods to customise it.

```rust
use lapin::{Connection, ConnectionProperties, Result};
# async fn example() -> Result<()> {
let props = ConnectionProperties::default()
    .with_connection_name("my-app".into())
    .enable_auto_recover();
let conn = Connection::connect("amqp://localhost", props).await?;
# Ok(())
# }
```

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

### Example Usage of basic_consume

Source: https://docs.rs/lapin/latest/lapin/struct.Consumer.html

Demonstrates how to create a consumer, receive messages, and acknowledge them.

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

fn main() -> Result<()> {
    let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
    let runtime = lapin::runtime::default_runtime().unwrap();

    runtime.clone().block_on(async move {
        let conn = Connection::connect_with_runtime(
            &addr,
            ConnectionProperties::default(),
            runtime,
        )
        .await?;
        let channel = conn.create_channel().await?;
        let mut consumer = channel
            .basic_consume(
                "hello".into(),
                "my_consumer".into(),
                BasicConsumeOptions::default(),
                FieldTable::default(),
            )
            .await?;

        while let Some(delivery) = consumer.next().await {
            let delivery = delivery.expect("error in consumer");
            delivery
                .ack(BasicAckOptions::default())
                .await?;
        }
        Ok(())
    })
}
```

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

### Handling connection start

Source: https://docs.rs/lapin/latest/src/lapin/channel.rs.html

This snippet shows the `on_connection_start_received` method, which processes the `connection::Start` method from the server. It validates the connection state and checks for supported authentication mechanisms and locales.

```rust
    fn on_connection_start_received(
        &self,
        method: protocol::connection::Start,
        step: ConnectionStep,
    ) -> Result<()> {
        trace!(?method, "Server sent connection::Start");

        let state = self.connection_status.state();
        if state != ConnectionState::Connecting {
            return self.connection_process_error(state, Some(step), None);
        }

        match step {
            ConnectionStep::ProtocolHeader(resolver, mut connection) => {
                let configuration = connection.configuration_mut();
                let auth_provider = configuration.auth_provider.clone();
                let mechanism = auth_provider.mechanism();
                let locale = configuration.amqp_locale.clone();

                if !method
                    .mechanisms
                    .to_string()
                    .split_whitespace()
                    .any(|m| m == mechanism.as_str())
                {
                    error!(%mechanism, "unsupported mechanism");
                }
                if !method
                    .locales
                    .to_string()
                    .split_whitespace()
                    .any(|l| l == locale.as_str())
                {

```

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

### Instrument Trait Implementation

Source: https://docs.rs/lapin/latest/lapin/protocol/connection/struct.Start.html

Blanket implementation of the Instrument trait.

```rust
fn instrument(self, span: Span) -> Instrumented<Self>
```

```rust
fn in_current_span(self) -> Instrumented<Self>
```

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

### Start Recovery

Source: https://docs.rs/lapin/latest/src/lapin/channel.rs.html

Initiates the channel recovery process, including reopening the channel and re-enabling confirms.

```rust
pub(crate) async fn start_recovery(&self) -> Result<()> {
        let topology = self
            .update_recovery()
            .ok_or_else(|| self.status.state_error("start_recovery"))?;

        // First, reopen the channel
        self.channel_open(self.clone()).await?;

        // Then, reenable confirm_select if needed
        if self.status.confirm() {
            self.confirm_select(ConfirmSelectOptions::default()).await?;
        }

        // Third, redeclare all exchanges
        for ex in &topology.exchanges {

```

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

### Consumer next() method example

Source: https://docs.rs/lapin/latest/src/lapin/consumer.rs.html

This snippet demonstrates how to use the `next()` method of the consumer and asserts the expected behavior when channels reach their limit.

```rust
let mut next = consumer.next();

assert_eq!(awoken_count.0.load(Ordering::SeqCst), 1);
assert_eq!(
    Pin::new(&mut next).poll(&mut cx),
    Poll::Ready(Some(Err(ErrorKind::ChannelsLimitReached.into())))
);
```

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

### Starting the Internal RPC Runner

Source: https://docs.rs/lapin/latest/src/lapin/internal_rpc.rs.html

Spawns a new task to run the internal RPC event loop.

```rust
    pub(crate) fn start(self, channels: Channels) {
        self.runtime.clone().spawn(self.run(channels));
    }

```

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

### Pointable Trait Implementation

Source: https://docs.rs/lapin/latest/lapin/protocol/connection/struct.Start.html

Blanket implementation of the Pointable trait.

```rust
const ALIGN: usize
```

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

### Basic AMQP Method Options

Source: https://docs.rs/lapin/latest/src/lapin/opt/rustwide/target/x86_64-unknown-linux-gnu/debug/build/lapin-0aa65c3f4b556b6e/out/channel.rs.html

This snippet shows the Rust struct definitions for options related to various basic AMQP methods like qos, consume, cancel, publish, deliver, get, ack, reject, recover-async, recover, and nack. Each struct contains boolean fields corresponding to specific flags for these methods.

```rust
1/// Options structs for every AMQP method that accepts flag arguments.
2pub mod options {
3  use super::*;
4
5/// Options for the `basic.qos` AMQP method.
6  #[derive(Copy, Clone, Debug, Default, PartialEq)]
7  pub struct BasicQosOptions {
8/// Apply quality-of-service settings globally to the entire connection rather than just this channel.
9    pub global: Boolean,
10}
11
12/// Options for the `basic.consume` AMQP method.
13  #[derive(Copy, Clone, Debug, Default, PartialEq)]
14  pub struct BasicConsumeOptions {
15/// Do not receive messages published by this connection on this consumer.
16    pub no_local: Boolean,
17/// Disable message acknowledgement; the server dequeues messages as soon as they are delivered.
18    pub no_ack: Boolean,
19/// Restrict access to the declaring connection; the entity is deleted when that connection closes.
20    pub exclusive: Boolean,
21/// Do not wait for a server confirmation; the operation is fire-and-forget.
22    pub nowait: Boolean,
23}
24
25/// Options for the `basic.cancel` AMQP method.
26  #[derive(Copy, Clone, Debug, Default, PartialEq)]
27  pub struct BasicCancelOptions {
28/// Do not wait for a server confirmation; the operation is fire-and-fire.
29    pub nowait: Boolean,
30}
31
32/// Options for the `basic.publish` AMQP method.
33  #[derive(Copy, Clone, Debug, Default, PartialEq)]
34  pub struct BasicPublishOptions {
35/// Return the message to the publisher if it cannot be routed to at least one queue.
36    pub mandatory: Boolean,
37/// Return the message to the publisher if no consumer is immediately available to receive it.
38    pub immediate: Boolean,
39}
40
41/// Options for the `basic.deliver` AMQP method.
42  #[derive(Copy, Clone, Debug, Default, PartialEq)]
43  pub struct BasicDeliverOptions {
44/// Indicates the message was previously delivered but not acknowledged.
45    pub redelivered: Boolean,
46}
47
48/// Options for the `basic.get` AMQP method.
49  #[derive(Copy, Clone, Debug, Default, PartialEq)]
50  pub struct BasicGetOptions {
51/// Disable message acknowledgement; the server dequeues messages as soon as they are delivered.
52    pub no_ack: Boolean,
53}
54
55/// Options for the `basic.get-ok` AMQP method.
56  #[derive(Copy, Clone, Debug, Default, PartialEq)]
57  pub struct BasicGetOkOptions {
58/// Indicates the message was previously delivered but not acknowledged.
59    pub redelivered: Boolean,
60}
61
62/// Options for the `basic.ack` AMQP method.
63  #[derive(Copy, Clone, Debug, Default, PartialEq)]
64  pub struct BasicAckOptions {
65/// Acknowledge or reject all outstanding deliveries up to and including this delivery tag.
66    pub multiple: Boolean,
67}
68
69/// Options for the `basic.reject` AMQP method.
70  #[derive(Copy, Clone, Debug, Default, PartialEq)]
71  pub struct BasicRejectOptions {
72/// Re-queue the message rather than discarding or dead-lettering it.
73    pub requeue: Boolean,
74}
75
76/// Options for the `basic.recover-async` AMQP method.
77  #[derive(Copy, Clone, Debug, Default, PartialEq)]
78  pub struct BasicRecoverAsyncOptions {
79/// Re-queue the message rather than discarding or dead-lettering it.
80    pub requeue: Boolean,
81}
82
83/// Options for the `basic.recover` AMQP method.
84  #[derive(Copy, Clone, Debug, Default, PartialEq)]
85  pub struct BasicRecoverOptions {
86/// Re-queue the message rather than discarding or dead-lettering it.
87    pub requeue: Boolean,
88}
89
90/// Options for the `basic.nack` AMQP method.
91  #[derive(Copy, Clone, Debug, Default, PartialEq)]
92  pub struct BasicNackOptions {
93/// Acknowledge or reject all outstanding deliveries up to and including this delivery tag.
94    pub multiple: Boolean,
95/// Re-queue the message rather than discarding or dead-lettering it.
96    pub requeue: Boolean,
97}
98
99/// Options for the `channel.flow` AMQP method.
100  #[derive(Copy, Clone, Debug, Default, PartialEq)]
101  pub struct ChannelFlowOptions {
102/// Enable or disable message flow on the channel.
103    pub active: Boolean,
104}
105
106/// Options for the `channel.flow-ok` AMQP method.
107  #[derive(Copy, Clone, Debug, Default, PartialEq)]
108  pub struct ChannelFlowOkOptions {
109/// Enable or disable message flow on the channel.
110    pub active: Boolean,
111}
112
113/// Options for the `access.request` AMQP method.
114  #[derive(Copy, Clone, Debug, Default, PartialEq)]
115  pub struct AccessRequestOptions {
116/// Restrict access to the declaring connection; the entity is deleted when that connection closes.
117    pub exclusive: Boolean,
118/// Verify that the exchange or queue exists without creating or modifying it.
119    pub passive: Boolean,
120/// Enable or disable message flow on the channel.
121    pub active: Boolean,
122/// Request write permission for the resource (access class only).
123    pub write: Boolean,
124}
```

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

### iter

Source: https://docs.rs/lapin/latest/lapin/type.Result.html

Example of using iter to get an iterator over the contained value if Ok.

```rust
let x: Result<u32, &str> = Ok(7);
assert_eq!(x.iter().next(), Some(&7));

let x: Result<u32, &str> = Err("nothing!");
assert_eq!(x.iter().next(), None);
```

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

### ConsumerStatusInner Structure and Methods

Source: https://docs.rs/lapin/latest/src/lapin/consumer_status.rs.html

The inner struct holding the actual state and delegate. Provides methods to get the state, get the delegate, set the delegate, start cancellation, and cancel the consumer.

```rust
#[derive(Default)]
pub(crate) struct ConsumerStatusInner {
    state: ConsumerState,
    delegate: Option<Arc<dyn ConsumerDelegate>>,
}

impl ConsumerStatusInner {
    pub(crate) fn state(&self) -> ConsumerState {
        self.state
    }

    pub(crate) fn delegate(&self) -> Option<Arc<dyn ConsumerDelegate>> {
        self.delegate.clone()
    }

    pub(crate) fn set_delegate(&mut self, delegate: Option<Arc<dyn ConsumerDelegate>>) {
        if self.state.is_active() {
            self.state = ConsumerState::ActiveWithDelegate;
            self.delegate = delegate;
        }
    }

    pub(crate) fn start_cancel(&mut self) {
        self.state = ConsumerState::Canceling;
        self.delegate = None;
    }

    pub(crate) fn cancel(&mut self) {
        self.state = ConsumerState::Canceled;
        self.delegate = None;
    }
}
```

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

### iter_mut

Source: https://docs.rs/lapin/latest/lapin/type.Result.html

Example of using iter_mut to get a mutable iterator over the contained value if Ok.

```rust
let mut x: Result<u32, &str> = Ok(7);
match x.iter_mut().next() {
    Some(v) => *v = 40,
    None => {},
}
assert_eq!(x, Ok(40));

let mut x: Result<u32, &str> = Err("nothing!");
assert_eq!(x.iter_mut().next(), None);
```

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

### IoLoop::ensure_setup method

Source: https://docs.rs/lapin/latest/src/lapin/io_loop.rs.html

Ensures that the I/O loop is set up correctly. If the status is Initial, it calls `finish_setup`. If Connected, it returns true. If Stop, it returns false.

```rust
fn ensure_setup(&mut self) -> Result<bool> {
    match self.status {
        Status::Initial => self.finish_setup(),
        Status::Connected => Ok(true),
        Status::Stop => Ok(false),
    }
}
```

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

### Struct Get

Source: https://docs.rs/lapin/latest/lapin/protocol/basic/struct.Get.html

Definition of the Get struct, which represents a GET AMQP method.

```rust
pub struct Get {
    pub queue: ShortString,
    pub no_ack: bool,
}
```

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

### and() - Examples

Source: https://docs.rs/lapin/latest/lapin/type.Result.html

Provides examples of the and() method, showing how it combines two Result values.

```rust
let x: Result<u32, &str> = Ok(2);
let y: Result<&str, &str> = Err("late error");
assert_eq!(x.and(y), Err("late error"));

let x: Result<u32, &str> = Err("early error");
let y: Result<&str, &str> = Ok("foo");
assert_eq!(x.and(y), Err("early error"));

let x: Result<u32, &str> = Err("not a 2");
let y: Result<&str, &str> = Err("late error");
assert_eq!(x.and(y), Err("not a 2"));

let x: Result<u32, &str> = Ok(2);
let y: Result<&str, &str> = Ok("different result type");
assert_eq!(x.and(y), Ok("different result type"));
```

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

### before_basic_publish

Source: https://docs.rs/lapin/latest/src/lapin/channel.rs.html

Prepares for a basic publish operation, registering a pending publisher confirm if publisher confirms are enabled.

```rust
fn before_basic_publish(&self) -> Option<PublisherConfirm> {
        if self.status.confirm() {
            Some(self.acknowledgements.register_pending())
        } else {
            None
        }
    }
```

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

### gen_start_ok

Source: https://docs.rs/lapin/latest/lapin/protocol/connection/fn.gen_start_ok.html

Serialize start-ok (Generated)

```rust
pub fn gen_start_ok<'a, W>(method: &'a StartOk) -> impl SerializeFn<W> + 'a
where
    W: Write + BackToTheBuffer + 'a,

```

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

### connect()

Source: https://docs.rs/lapin/latest/lapin/type.DefaultConnectionBuilder.html

Establish the connection with the configured parameters.

```Rust
pub async fn connect(&self) -> Result<Connection>
```

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

### Any::type_id

Source: https://docs.rs/lapin/latest/lapin/auth/struct.TokenAuthProvider.html

Gets the TypeId of self.

```rust
fn type_id(&self) -> TypeId
```

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

### basic_consume method

Source: https://docs.rs/lapin/latest/lapin/struct.Channel.html

Start a message consumer on `queue`.

```rust
pub async fn basic_consume(
 &self,
 queue: ShortString,
 consumer_tag: ShortString,
 options: BasicConsumeOptions,
 arguments: FieldTable,
) -> Result<Consumer>
```

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

### Start Message Consumer

Source: https://docs.rs/lapin/latest/src/lapin/channel.rs.html

Starts a message consumer on a specified queue. Returns a Consumer stream that yields Delivery values. The consumer_tag uniquely identifies this consumer on the channel; pass an empty string to let the server generate one. See Consumer for acknowledgement and prefetch documentation.

```Rust
    /// Start a message consumer on `queue`.
    ///
    /// Returns a [`Consumer`] stream that yields [`Delivery`] values.
    /// `consumer_tag` uniquely identifies this consumer on the channel;
    /// pass an empty string to let the server generate one.
    ///
    /// See [`Consumer`] for acknowledgement and prefetch documentation.

```

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

### IntoIterator for Result

Source: https://docs.rs/lapin/latest/lapin/type.Result.html

Examples demonstrating how to convert a Result into an iterator.

```rust
let x: Result<u32, &str> = Ok(5);
let v: Vec<u32> = x.into_iter().collect();
assert_eq!(v, [5]);
```

```rust
let x: Result<u32, &str> = Err("nothing!");
let v: Vec<u32> = x.into_iter().collect();
assert_eq!(v, []);
```

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

### Wait for recovery

Source: https://docs.rs/lapin/latest/index.html

Example of waiting for connection recovery after an error.

```rust
channel.wait_for_recovery(error).await?;
```

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

### DeliveryCause Enum

Source: https://docs.rs/lapin/latest/src/lapin/channel_receiver_state.rs.html

Defines the possible causes for delivery: Consume, Get, or Return.

```rust
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum DeliveryCause {
    Consume(ShortString),
    Get,
    Return,
}

```

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

### unwrap_err() - Example

Source: https://docs.rs/lapin/latest/lapin/type.Result.html

Shows unwrap_err() returning the contained Err value.

```rust
let x: Result<u32, &str> = Err("emergency failure");
assert_eq!(x.unwrap_err(), "emergency failure");
```

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

### fn take

Source: https://docs.rs/lapin/latest/lapin/enum.AsyncTcpStream.html

Creates an adapter which will read at most `limit` bytes from it.

```rust
fn take(self, limit: u64) -> Take<Self>
where Self: Sized,
```

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

### map_err

Source: https://docs.rs/lapin/latest/lapin/type.Result.html

Example of using map_err to transform the error type of a Result.

```rust
fn stringify(x: u32) -> String { format!("error code: {x}") }

let x: Result<u32, u32> = Ok(2);
assert_eq!(x.map_err(stringify), Ok(2));

let x: Result<u32, u32> = Err(13);
assert_eq!(x.map_err(stringify), Err("error code: 13".to_string()));
```

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

### Receive Access Request OK Method

Source: https://docs.rs/lapin/latest/src/lapin/opt/rustwide/target/x86_64-unknown-linux-gnu/debug/build/lapin-0aa65c3f4b556b6e/out/channel.rs.html

Handles the confirmation of an access request.

```rust
fn receive_access_request_ok(&self, method: protocol::access::RequestOk) -> Result<()> {
    if !self.status.can_receive_messages() {
return Err(self.status.state_error("access.request-ok"));
    }

    match self.frames.find_expected_reply(self.id, |reply| matches!(&reply.0, Reply::AccessRequestOk(..))){
      Some(Reply::AccessRequestOk(resolver)) => {
        let res =self.on_access_request_ok_received(method)
;
        resolver.complete(res.clone());
        res
},
      unexpected => {
        self.handle_invalid_contents(format!("unexpected access request-ok received on channel {}, was awaiting for {:?}", self.id, unexpected), method.get_amqp_class_id(), method.get_amqp_method_id())
      },
    }
}
```

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

### Wait for recovery

Source: https://docs.rs/lapin/latest/src/lapin/lib.rs.html

Example of waiting for connection recovery after an error.

```rust
# use lapin::{Channel, Error, Result};
# async fn example(channel: Channel, error: Error) -> Result<()> {
channel.wait_for_recovery(error).await?;
# Ok(())
# }
```

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

### Connection Open

Source: https://docs.rs/lapin/latest/src/lapin/opt/rustwide/target/x86_64-unknown-linux-gnu/debug/build/lapin-0aa65c3f4b556b6e/out/channel.rs.html

Requests to open a virtual host, completing the connection handshake.

```rust
797/// Client request to open a virtual host, completing the connection handshake.
798pub (crate) async fn connection_open(&self, virtual_host: ShortString, connection: Box<Connection>, conn_resolver: PromiseResolver<Connection>) -> Result<()> {
799let (promise, resolver) = Promise::new("connection.open");
800let reply = Reply::ConnectionOpenOk(resolver.clone(), connection, conn_resolver);
801let method = AMQPClass::Connection(protocol::connection::AMQPMethod::Open (protocol::connection::Open {
802virtual_host,
803}));
804
805    self.send_method_frame(method, Box::new(resolver.clone()), Some(ExpectedReply(reply, Box::new(resolver))), None);
806promise.await
```

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

### Basic Get

Source: https://docs.rs/lapin/latest/src/lapin/channel.rs.html

Asynchronously fetches a single message from a queue. Returns None if the queue is empty.

```rust
pub async fn basic_get(
        &self,
        queue: ShortString,
        options: BasicGetOptions,
    ) -> Result<Option<BasicGetMessage>> {
        self.do_basic_get(queue, options, None).await
    }
```

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

### Connection Tune OK

Source: https://docs.rs/lapin/latest/src/lapin/opt/rustwide/target/x86_64-unknown-linux-gnu/debug/build/lapin-0aa65c3f4b556b6e/out/channel.rs.html

Confirms negotiated connection parameters like channel max, frame max, and heartbeat.

```rust
785async fn connection_tune_ok(&self, channel_max: ShortUInt, frame_max: LongUInt, heartbeat: ShortUInt) -> Result<()> {
786let (promise, resolver) = Promise::new("connection.tune-ok");
787let method = AMQPClass::Connection(protocol::connection::AMQPMethod::TuneOk (protocol::connection::TuneOk {
788channel_max,
789frame_max,
790heartbeat,
791}));
792
793    self.send_method_frame(method, Box::new(resolver.clone()), None, Some(resolver));
794promise.await
795}
```

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

### TokenAuthProvider::auth_starter

Source: https://docs.rs/lapin/latest/lapin/auth/struct.TokenAuthProvider.html

Provides the initial data to the RabbitMQ server for authentication.

```rust
fn auth_starter(&self) -> Result<LongString, String>
```

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

### Connection Tune OK and Open

Source: https://docs.rs/lapin/latest/src/lapin/channel.rs.html

Spawns an asynchronous task to send connection tuning parameters and open the connection.

```rust
let configuration = self.configuration.clone();
let vhost = self.connection_status.vhost();
self.internal_rpc.spawn(async move {
    channel
        .connection_tune_ok(
            configuration.channel_max(),
            configuration.frame_max(),
            configuration.heartbeat(),
        )
        .await?;
    channel
        .connection_open(vhost, Box::new(connection), resolver)
        .await
});
Ok(())
```

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

### IntoEither Trait Implementation

Source: https://docs.rs/lapin/latest/lapin/protocol/connection/struct.Start.html

Blanket implementation of the IntoEither trait.

```rust
fn into_either(self, into_left: bool) -> Either<Self, Self>
```

```rust
fn into_either_with<F>(self, into_left: F) -> Either<Self, Self>
```

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

### new()

Source: https://docs.rs/lapin/latest/lapin/type.DefaultConnectionBuilder.html

Create a builder using the crate’s default async runtime.

```Rust
pub fn new() -> Result<Self>
```

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

### Connect using a pre-parsed AMQPUri, explicit runtime, and TLS configuration.

Source: https://docs.rs/lapin/latest/src/lapin/connection.rs.html

This function combines `connect_uri` and `connect_with_config` for establishing a connection with specific TLS settings.

```rust
pub async fn connect_uri_with_config<RK: RuntimeKit + Send + Sync + Clone + 'static>(
    uri: AMQPUri,
    options: ConnectionProperties,
    config: OwnedTLSConfig,
    runtime: Runtime<RK>,
) -> Result<Self> {
    uri.connect_with_config(options, config, runtime).await
}
```

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

### Enable automatic connection recovery

Source: https://docs.rs/lapin/latest/index.html

Example of enabling automatic connection recovery in ConnectionProperties.

```rust
use lapin::ConnectionProperties;

let props = ConnectionProperties::default().enable_auto_recover();
// then pass `props` to Connection::connect(…)
```

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

### Connect Implementation for &str

Source: https://docs.rs/lapin/latest/src/lapin/connection.rs.html

Implements the `connect_with_config` method for string slices (`&str`) by parsing them into `AMQPUri` first.

```rust
impl Connect for &str {
    async fn connect_with_config<RK: RuntimeKit + Send + Sync + Clone + 'static>(
        self,
        options: ConnectionProperties,
        config: OwnedTLSConfig,
        runtime: Runtime<RK>,
    ) -> Result<Connection> {
        match self.parse::<AMQPUri>() {
            Ok(uri) => uri.connect_with_config(options, config, runtime).await,
            Err(err) => Err(Error::other(err)),
        }
    }
}
```

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

### Buffer data availability and space

Source: https://docs.rs/lapin/latest/src/lapin/buffer.rs.html

Methods to get the amount of available data and available space in the buffer.

```rust
    pub(crate) fn available_data(&self) -> usize {
        self.available_data
    }

    pub(crate) fn available_space(&self) -> usize {
        self.capacity - self.available_data
    }

```

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

### gen_get Function Signature

Source: https://docs.rs/lapin/latest/lapin/protocol/basic/fn.gen_get.html

The function signature for gen_get, which serializes a 'Get' method.

```rust
pub fn gen_get<'a, W>(method: &'a Get) -> impl SerializeFn<W> + 'a
where
    W: Write + BackToTheBuffer + 'a,

```

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

### new_with_runtime()

Source: https://docs.rs/lapin/latest/lapin/type.DefaultConnectionBuilder.html

Create a builder with a custom runtime.

```Rust
pub fn new_with_runtime(runtime: Runtime<RK>) -> Self
```

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

### create_channel Method

Source: https://docs.rs/lapin/latest/src/lapin/internal_rpc.rs.html

Initiates the creation of a new AMQP channel. It takes a ConnectionCloser and returns a Promise that resolves to a Channel upon successful creation.

```rust
pub(crate) async fn create_channel(
    &self,
    connection_closer: Arc<ConnectionCloser>,
) -> Result<Channel> {
    let (promise, resolver) = Promise::new("channel.create");
    self.send(InternalCommand::CreateChannel(connection_closer, resolver));
    promise.await
}

```

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

### FromIterator for Result

Source: https://docs.rs/lapin/latest/lapin/type.Result.html

Example of collecting an iterator of Results into a single Result, handling potential errors.

```rust
let v = vec![1, 2];
let res: Result<Vec<u32>, &'static str> = v.iter().map(|x: &u32| 
    x.checked_add(1).ok_or("Overflow!")
).collect();
assert_eq!(res, Ok(vec![2, 3]));
```

```rust
let v = vec![1, 2, 0];
let res: Result<Vec<u32>, &'static str> = v.iter().map(|x: &u32| 
    x.checked_sub(1).ok_or("Underflow!")
).collect();
assert_eq!(res, Err("Underflow!"));
```

```rust
let v = vec![3, 2, 1, 10];
let mut shared = 0;
let res: Result<Vec<u32>, &'static str> = v.iter().map(|x: &u32| {
    shared += x;
    x.checked_sub(2).ok_or("Underflow!")
}).collect();
assert_eq!(res, Err("Underflow!"));
assert_eq!(shared, 6);
```

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

### into_err() - Example

Source: https://docs.rs/lapin/latest/lapin/type.Result.html

Demonstrates the experimental into_err() method, which returns the Err value without panicking.

```rust
fn only_bad_news() -> Result<!, String> {
    Err("Oops, it failed".into())
}

let error: String = only_bad_news().into_err();
println!("{error}");
```

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

### with_uri()

Source: https://docs.rs/lapin/latest/lapin/type.DefaultConnectionBuilder.html

Set the broker URI from a pre-parsed AMQPUri.

```Rust
pub fn with_uri(self, uri: AMQPUri) -> Self
```

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

### basic_qos method

Source: https://docs.rs/lapin/latest/src/lapin/opt/rustwide/target/x86_64-unknown-linux-gnu/debug/build/lapin-0aa65c3f4b556b6e/out/channel.rs.html

Sets the Quality of Service for the channel, controlling message prefetching and global limits.

```Rust
pub async fn basic_qos(&self, prefetch_count: ShortUInt, options: BasicQosOptions) -> Result<()> {
    if !self.status.connected() {
return Err(self.status.state_error("basic.qos"));
    }

let BasicQosOptions {
global,
} = options;
let (promise, resolver) = Promise::new("basic.qos");
let reply = Reply::BasicQosOk(resolver.clone());
let method = AMQPClass::Basic(protocol::basic::AMQPMethod::Qos (protocol::basic::Qos {
prefetch_count,
global,
}));

    self.send_method_frame(method, Box::new(resolver.clone()), Some(ExpectedReply(reply, Box::new(resolver))), None);
promise.await
}
```

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

### receive_basic_qos_ok function

Source: https://docs.rs/lapin/latest/src/lapin/opt/rustwide/target/x86_64-unknown-linux-gnu/debug/build/lapin-0aa65c3f4b556b6e/out/channel.rs.html

Handles the server's response to a basic_qos request.

```Rust
fn receive_basic_qos_ok(&self, method: protocol::basic::QosOk) -> Result<()> {
    if !self.status.can_receive_messages() {
return Err(self.status.state_error("basic.qos-ok"));
    }

    match self.frames.find_expected_reply(self.id, |reply| matches!(&reply.0, Reply::BasicQosOk(..))){
      Some(Reply::BasicQosOk(resolver)) => {
        let res =        Ok(())
;
        resolver.complete(res.clone());
        res
},
      unexpected => {
        self.handle_invalid_contents(format!("unexpected basic qos-ok received on channel {}, was awaiting for {:?}", self.id, unexpected), method.get_amqp_class_id(), method.get_amqp_method_id())
      },
    }
}
```
Tokens:
Raw