add a way to wait for channel recovery

Signed-off-by: Marc-Antoine Perennou <[email protected]>

Author: Keruspe

src/channel.rs

@@ -566,7 +566,7 @@ impl Channel {
 
     fn next_expected_close_ok_reply(&self) -> Option<Reply> {
         self.frames
-            .next_expected_close_ok_reply(self.id, Error::InvalidChannelState(ChannelState::Closed))
+            .next_expected_close_ok_reply(self.id, Error::InvalidChannelState(ChannelState::Closed, None))
     }
 
     fn before_channel_close(&self) {
@@ -582,7 +582,7 @@ impl Channel {
                 self.set_closed(
                     error
                         .clone()
-                        .unwrap_or(Error::InvalidChannelState(ChannelState::Closing)),
+                        .unwrap_or(Error::InvalidChannelState(ChannelState::Closing, None)),
                 );
                 if let Some(error) = error {
                     self.error_handler.on_error(error);
@@ -936,7 +936,7 @@ impl Channel {
     }
 
     fn on_channel_close_ok_received(&self) -> Result<()> {
-        self.set_closed(Error::InvalidChannelState(ChannelState::Closed));
+        self.set_closed(Error::InvalidChannelState(ChannelState::Closed, None));
         Ok(())
     }
 

src/channel_recovery_context.rs

@@ -1,5 +1,6 @@
 use crate::{
     frames::{ExpectedReply, Frames},
+    notifier::Notifier,
     Error,
 };
 
@@ -8,16 +9,22 @@ use std::collections::VecDeque;
 pub(crate) struct ChannelRecoveryContext {
     cause: Error,
     expected_replies: Option<VecDeque<ExpectedReply>>,
+    notifier: Notifier,
 }
 
 impl ChannelRecoveryContext {
     pub(crate) fn new(cause: Error) -> Self {
         Self {
             cause,
             expected_replies: None,
+            notifier: Notifier::default(),
         }
     }
 
+    pub(crate) fn notifier(&self) -> Notifier {
+        self.notifier.clone()
+    }
+
     pub(crate) fn set_expected_replies(
         &mut self,
         expected_replies: Option<VecDeque<ExpectedReply>>,
@@ -26,6 +33,7 @@ impl ChannelRecoveryContext {
     }
 
     pub(crate) fn finalize_recovery(self) {
+        self.notifier.notify_all();
         if let Some(replies) = self.expected_replies {
             Frames::cancel_expected_replies(replies, self.cause);
         }

src/channel_status.rs

@@ -1,6 +1,7 @@
 use crate::{
     channel_receiver_state::{ChannelReceiverStates, DeliveryCause},
     channel_recovery_context::ChannelRecoveryContext,
+    notifier::Notifier,
     types::{ChannelId, Identifier, PayloadSize},
     Error, Result,
 };
@@ -67,6 +68,11 @@ impl ChannelStatus {
         self.0.lock().state = state;
     }
 
+    pub fn state_error(&self) -> Error {
+        let inner = self.0.lock();
+        Error::InvalidChannelState(inner.state.clone(), inner.notifier())
+    }
+
     pub(crate) fn set_reconnecting(&self, error: Error) {
         let mut inner = self.0.lock();
         inner.state = ChannelState::Reconnecting;
@@ -193,4 +199,8 @@ impl Inner {
             ctx.finalize_recovery();
         }
     }
+
+    fn notifier(&self) -> Option<Notifier> {
+        Some(self.recovery_context.as_ref()?.notifier())
+    }
 }

src/error.rs

@@ -1,6 +1,6 @@
 use crate::{
-    channel_status::ChannelState, connection_status::ConnectionState, protocol::AMQPError,
-    types::ChannelId,
+    channel_status::ChannelState, connection_status::ConnectionState, notifier::Notifier,
+    protocol::AMQPError, types::ChannelId,
 };
 use amq_protocol::{
     frame::{GenError, ParserError, ProtocolVersion},
@@ -22,7 +22,7 @@ pub enum Error {
     InvalidProtocolVersion(ProtocolVersion),
 
     InvalidChannel(ChannelId),
-    InvalidChannelState(ChannelState),
+    InvalidChannelState(ChannelState, Option<Notifier>),
     InvalidConnectionState(ConnectionState),
 
     IOError(Arc<io::Error>),
@@ -84,7 +84,7 @@ impl fmt::Display for Error {
             }
 
             Error::InvalidChannel(channel) => write!(f, "invalid channel: {}", channel),
-            Error::InvalidChannelState(state) => write!(f, "invalid channel state: {:?}", state),
+            Error::InvalidChannelState(state, _) => write!(f, "invalid channel state: {:?}", state),
             Error::InvalidConnectionState(state) => {
                 write!(f, "invalid connection state: {:?}", state)
             }
@@ -144,7 +144,7 @@ impl PartialEq for Error {
             }
 
             (InvalidChannel(left_inner), InvalidChannel(right_inner)) => left_inner == right_inner,
-            (InvalidChannelState(left_inner), InvalidChannelState(right_inner)) => {
+            (InvalidChannelState(left_inner, _), InvalidChannelState(right_inner, _)) => {
                 left_inner == right_inner
             }
             (InvalidConnectionState(left_inner), InvalidConnectionState(right_inner)) => {