### 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 { 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, auth_provider: Arc) -> 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 + '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 { 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 = 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 { 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::>(); 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 ``` ```rust fn in_current_span(self) -> Instrumented ``` -------------------------------- ### 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 = Ok(7); assert_eq!(x.iter().next(), Some(&7)); let x: Result = 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>, } impl ConsumerStatusInner { pub(crate) fn state(&self) -> ConsumerState { self.state } pub(crate) fn delegate(&self) -> Option> { self.delegate.clone() } pub(crate) fn set_delegate(&mut self, delegate: Option>) { 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 = Ok(7); match x.iter_mut().next() { Some(v) => *v = 40, None => {}, } assert_eq!(x, Ok(40)); let mut x: Result = 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 { 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 = Ok(2); let y: Result<&str, &str> = Err("late error"); assert_eq!(x.and(y), Err("late error")); let x: Result = Err("early error"); let y: Result<&str, &str> = Ok("foo"); assert_eq!(x.and(y), Err("early error")); let x: Result = Err("not a 2"); let y: Result<&str, &str> = Err("late error"); assert_eq!(x.and(y), Err("not a 2")); let x: Result = 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 { 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 + '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 ``` -------------------------------- ### 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 ``` -------------------------------- ### 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 = Ok(5); let v: Vec = x.into_iter().collect(); assert_eq!(v, [5]); ``` ```rust let x: Result = Err("nothing!"); let v: Vec = 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 = 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 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 = Ok(2); assert_eq!(x.map_err(stringify), Ok(2)); let x: Result = 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, conn_resolver: PromiseResolver) -> 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> { 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 ``` -------------------------------- ### 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 ``` ```rust fn into_either_with(self, into_left: F) -> Either ``` -------------------------------- ### 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 ``` -------------------------------- ### 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( uri: AMQPUri, options: ConnectionProperties, config: OwnedTLSConfig, runtime: Runtime, ) -> Result { 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(…) ``` -------------------------------- ### 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 + 'a where W: Write + BackToTheBuffer + 'a, ``` -------------------------------- ### 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( self, options: ConnectionProperties, config: OwnedTLSConfig, runtime: Runtime, ) -> Result { match self.parse::() { Ok(uri) => uri.connect_with_config(options, config, runtime).await, Err(err) => Err(Error::other(err)), } } } ``` -------------------------------- ### 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) -> 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, ) -> Result { 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, &'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, &'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, &'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 { 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()) }, } } ```