use lapin::{ message::DeliveryResult, options::*, publisher_confirm::Confirmation, types::FieldTable, BasicProperties, Connection, ConnectionProperties, Result, }; use lapin_async_global_executor::*; use tracing::info; fn main() -> Result<()> { if std::env::var("RUST_LOG").is_err() { std::env::set_var("RUST_LOG", "info"); } tracing_subscriber::fmt::init(); let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into()); async_global_executor::block_on(async { let conn = Connection::connect( &addr, ConnectionProperties::default().with_async_global_executor(), ) .await?; info!("CONNECTED"); let channel_a = conn.create_channel().await?; let channel_b = conn.create_channel().await?; let queue = channel_a .queue_declare( "hello", QueueDeclareOptions::default(), FieldTable::default(), ) .await?; info!(?queue, "Declared queue"); let consumer = channel_b .basic_consume( "hello", "my_consumer", BasicConsumeOptions::default(), FieldTable::default(), ) .await?; consumer.set_delegate(move |delivery: DeliveryResult| async move { let delivery = delivery.expect("error caught in in consumer"); if let Some(delivery) = delivery { delivery .ack(BasicAckOptions::default()) .await .expect("failed to ack"); } }); let payload = b"Hello world!"; loop { let confirm = channel_a .basic_publish( "", "hello", BasicPublishOptions::default(), payload, BasicProperties::default(), ) .await? .await?; assert_eq!(confirm, Confirmation::NotRequested); } }) }