Channel recovery

Cargo.toml

@@ -75,6 +75,7 @@ waker-fn = "^1.1"
 
 [dev-dependencies]
 async-global-executor = "^2.0"
+async-io = "^2.0"
 futures-lite = "^2.0"
 serde_json = "^1.0"
 waker-fn = "^1.1"

examples/c.rs

@@ -0,0 +1,58 @@
+use futures_lite::StreamExt;
+use lapin::{options::*, types::FieldTable, Connection, ConnectionProperties};
+use tracing::info;
+
+fn main() {
+    if std::env::var("RUST_LOG").is_err() {
+        unsafe { std::env::set_var("RUST_LOG", "info") };
+    }
+
+    tracing_subscriber::fmt::init();
+
+    let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
+
+    async_global_executor::block_on(async {
+        let conn = Connection::connect(&addr, ConnectionProperties::default())
+            .await
+            .expect("connection error");
+
+        info!("CONNECTED");
+
+        //receive channel
+        let channel = conn.create_channel().await.expect("create_channel");
+        info!(state=?conn.status().state());
+
+        let queue = channel
+            .queue_declare(
+                "hello-recover",
+                QueueDeclareOptions::default(),
+                FieldTable::default(),
+            )
+            .await
+            .expect("queue_declare");
+        info!(state=?conn.status().state());
+        info!(?queue, "Declared queue");
+
+        info!("will consume");
+        let mut consumer = channel
+            .basic_consume(
+                "hello-recover",
+                "my_consumer",
+                BasicConsumeOptions::default(),
+                FieldTable::default(),
+            )
+            .await
+            .expect("basic_consume");
+        info!(state=?conn.status().state());
+
+        while let Some(delivery) = consumer.next().await {
+            info!(message=?delivery, "received message");
+            if let Ok(delivery) = delivery {
+                delivery
+                    .ack(BasicAckOptions::default())
+                    .await
+                    .expect("basic_ack");
+            }
+        }
+    })
+}

examples/p.rs

@@ -0,0 +1,100 @@
+use lapin::{
+    options::*, types::FieldTable, BasicProperties, Connection, ConnectionProperties,
+};
+use tracing::info;
+
+fn main() {
+    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());
+    let recovery_config = lapin::experimental::RecoveryConfig {
+        auto_recover_channels: true,
+    };
+
+    async_global_executor::block_on(async {
+        let conn = Connection::connect(
+            &addr,
+            ConnectionProperties::default().with_experimental_recovery_config(recovery_config),
+        )
+        .await
+        .expect("connection error");
+
+        info!("CONNECTED");
+
+        let channel1 = conn.create_channel().await.expect("create_channel");
+        channel1
+            .confirm_select(ConfirmSelectOptions::default())
+            .await
+            .expect("confirm_select");
+        channel1
+            .queue_declare(
+                "hello-recover",
+                QueueDeclareOptions::default(),
+                FieldTable::default(),
+            )
+            .await
+            .expect("queue_declare");
+
+        let ch = channel1.clone();
+        async_global_executor::spawn(async move {
+            loop {
+                async_io::Timer::after(std::time::Duration::from_secs(1)).await;
+                info!("Trigger failure");
+                assert!(ch
+                    .queue_declare(
+                        "fake queue",
+                        QueueDeclareOptions {
+                            passive: true,
+                            ..QueueDeclareOptions::default()
+                        },
+                        FieldTable::default(),
+                    )
+                    .await
+                    .is_err());
+            }
+        })
+        .detach();
+
+        let mut published = 0;
+        let mut errors = 0;
+        info!("will publish");
+        loop {
+            let res = channel1
+                .basic_publish(
+                    "",
+                    "recover-test",
+                    BasicPublishOptions::default(),
+                    b"before",
+                    BasicProperties::default(),
+                )
+                .await;
+            let res = if let Ok(res) = res {
+                res.await.map(|_| ())
+            } else {
+                res.map(|_| ())
+            };
+            match res {
+                Ok(()) => {
+                    println!("GOT OK");
+                    published += 1;
+                }
+                Err(err) => {
+                    println!("GOT ERROR");
+                    let (soft, notifier) = err.is_amqp_soft_error();
+                    if !soft {
+                        panic!("{}", err);
+                    }
+                    errors += 1;
+                    if let Some(notifier) = notifier {
+                        notifier.await
+                    }
+                }
+            }
+            println!("Published {} with {} errors", published, errors);
+        }
+    });
+}

examples/t.rs

@@ -0,0 +1,122 @@
+use lapin::{
+    message::DeliveryResult, options::*, publisher_confirm::Confirmation, types::FieldTable,
+    BasicProperties, Connection, ConnectionProperties,
+};
+use tracing::info;
+
+fn main() {
+    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());
+    let recovery_config = lapin::experimental::RecoveryConfig {
+        auto_recover_channels: true,
+    };
+
+    async_global_executor::block_on(async {
+        let conn = Connection::connect(
+            &addr,
+            ConnectionProperties::default().with_experimental_recovery_config(recovery_config),
+        )
+        .await
+        .expect("connection error");
+
+        info!("CONNECTED");
+
+        {
+            let channel1 = conn.create_channel().await.expect("create_channel");
+            let channel2 = conn.create_channel().await.expect("create_channel");
+            channel1
+                .confirm_select(ConfirmSelectOptions::default())
+                .await
+                .expect("confirm_select");
+            channel1
+                .queue_declare(
+                    "recover-test",
+                    QueueDeclareOptions::default(),
+                    FieldTable::default(),
+                )
+                .await
+                .expect("queue_declare");
+
+            info!("will consume");
+            let channel = channel2.clone();
+            channel2
+                .basic_consume(
+                    "recover-test",
+                    "my_consumer",
+                    BasicConsumeOptions::default(),
+                    FieldTable::default(),
+                )
+                .await
+                .expect("basic_consume")
+                .set_delegate(move |delivery: DeliveryResult| {
+                    let channel = channel.clone();
+                    async move {
+                        info!(message=?delivery, "received message");
+                        if let Ok(Some(delivery)) = delivery {
+                            delivery
+                                .ack(BasicAckOptions::default())
+                                .await
+                                .expect("basic_ack");
+                            if &delivery.data[..] == b"after" {
+                                channel
+                                    .basic_cancel("my_consumer", BasicCancelOptions::default())
+                                    .await
+                                    .expect("basic_cancel");
+                            }
+                        }
+                    }
+                });
+
+            info!("will publish");
+            let confirm = channel1
+                .basic_publish(
+                    "",
+                    "recover-test",
+                    BasicPublishOptions::default(),
+                    b"before",
+                    BasicProperties::default(),
+                )
+                .await
+                .expect("basic_publish")
+                .await
+                .expect("publisher-confirms");
+            assert_eq!(confirm, Confirmation::Ack(None));
+
+            info!("before fail");
+            assert!(channel1
+                .queue_declare(
+                    "fake queue",
+                    QueueDeclareOptions {
+                        passive: true,
+                        ..QueueDeclareOptions::default()
+                    },
+                    FieldTable::default(),
+                )
+                .await
+                .is_err());
+            info!("after fail");
+
+            info!("publish after");
+            let confirm = channel1
+                .basic_publish(
+                    "",
+                    "recover-test",
+                    BasicPublishOptions::default(),
+                    b"after",
+                    BasicProperties::default(),
+                )
+                .await
+                .expect("basic_publish")
+                .await
+                .expect("publisher-confirms");
+            assert_eq!(confirm, Confirmation::Ack(None));
+        }
+
+        conn.run().expect("conn.run");
+    });
+}

src/acker.rs

@@ -78,10 +78,13 @@ impl Acker {
 
     async fn rpc<F: Fn(&InternalRPCHandle, PromiseResolver<()>)>(&self, f: F) -> Result<()> {
         if self.used.swap(true, Ordering::SeqCst) {
-            return Err(Error::ProtocolError(AMQPError::new(
-                AMQPSoftError::PRECONDITIONFAILED.into(),
-                "Attempted to use an already used Acker".into(),
-            )));
+            return Err(Error::ProtocolError(
+                AMQPError::new(
+                    AMQPSoftError::PRECONDITIONFAILED.into(),
+                    "Attempted to use an already used Acker".into(),
+                ),
+                None,
+            ));
         }
         if let Some(error) = self.error.as_ref() {
             error.check()?;

src/acknowledgement.rs

@@ -62,6 +62,10 @@ impl Acknowledgements {
     pub(crate) fn on_channel_error(&self, error: Error) {
         self.0.lock().on_channel_error(error);
     }
+
+    pub(crate) fn reset(&self, error: Error) {
+        self.0.lock().reset(error);
+    }
 }
 
 impl fmt::Debug for Acknowledgements {
@@ -174,4 +178,9 @@ impl Inner {
             }
         }
     }
+
+    fn reset(&mut self, error: Error) {
+        self.delivery_tag = IdSequence::new(false);
+        self.on_channel_error(error);
+    }
 }