propagate recovery_config in each Channel

Signed-off-by: Marc-Antoine Perennou <Marc-Antoine@Perennou.com>
amqp-rs · Aug 23, 2024 · 17ba64c · 17ba64c
1 parent fbb273b
commit 17ba64c
Show file tree

Hide file tree

Showing 3 changed files with 19 additions and 3 deletions.
diff --git a/src/channel.rs b/src/channel.rs
@@ -10,6 +10,7 @@ use crate::{
     consumer::Consumer,
     consumers::Consumers,
     error_handler::ErrorHandler,
+    experimental::RecoveryConfig,
     frames::{ExpectedReply, Frames},
     internal_rpc::InternalRPCHandle,
     message::{BasicGetMessage, BasicReturnMessage, Delivery},
@@ -59,6 +60,7 @@ pub struct Channel {
     executor: Arc<dyn FullExecutor + Send + Sync>,
     channel_closer: Option<Arc<ChannelCloser>>,
     connection_closer: Option<Arc<ConnectionCloser>>,
+    recovery_config: RecoveryConfig,
 }
 
 impl PartialEq for Channel {
@@ -94,6 +96,7 @@ impl Channel {
         frames: Frames,
         executor: Arc<dyn FullExecutor + Send + Sync>,
         connection_closer: Option<Arc<ConnectionCloser>>,
+        recovery_config: RecoveryConfig,
     ) -> Channel {
         let returned_messages = ReturnedMessages::default();
         let status = ChannelStatus::default();
@@ -106,7 +109,7 @@ impl Channel {
                 internal_rpc.clone(),
             )))
         };
-        Channel {
+        Self {
             id: channel_id,
             configuration,
             status,
@@ -124,6 +127,7 @@ impl Channel {
             executor,
             channel_closer,
             connection_closer,
+            recovery_config,
         }
     }
 
@@ -267,6 +271,7 @@ impl Channel {
             executor: self.executor.clone(),
             channel_closer: None,
             connection_closer: self.connection_closer.clone(),
+            recovery_config: self.recovery_config.clone(),
         }
     }
 

diff --git a/src/channels.rs b/src/channels.rs
@@ -1,6 +1,7 @@
 use crate::{
     connection_closer::ConnectionCloser,
     error_handler::ErrorHandler,
+    experimental::RecoveryConfig,
     frames::Frames,
     id_sequence::IdSequence,
     internal_rpc::InternalRPCHandle,
@@ -38,9 +39,10 @@ impl Channels {
         internal_rpc: InternalRPCHandle,
         frames: Frames,
         executor: Arc<dyn FullExecutor + Send + Sync>,
+        recovery_config: RecoveryConfig,
     ) -> Self {
         Self {
-            inner: Arc::new(Mutex::new(Inner::new(configuration, waker))),
+            inner: Arc::new(Mutex::new(Inner::new(configuration, waker, recovery_config))),
             connection_status,
             global_registry,
             internal_rpc,
@@ -296,15 +298,17 @@ struct Inner {
     channel_id: IdSequence<ChannelId>,
     configuration: Configuration,
     waker: SocketStateHandle,
+    recovery_config: RecoveryConfig,
 }
 
 impl Inner {
-    fn new(configuration: Configuration, waker: SocketStateHandle) -> Self {
+    fn new(configuration: Configuration, waker: SocketStateHandle, recovery_config: RecoveryConfig) -> Self {
         Self {
             channels: HashMap::default(),
             channel_id: IdSequence::new(false),
             configuration,
             waker,
+            recovery_config,
         }
     }
 
@@ -329,6 +333,7 @@ impl Inner {
             frames,
             executor,
             connection_closer,
+            self.recovery_config.clone(),
         );
         self.channels.insert(id, channel.clone_internal());
         channel

diff --git a/src/connection.rs b/src/connection.rs
@@ -5,6 +5,7 @@ use crate::{
     connection_closer::ConnectionCloser,
     connection_properties::ConnectionProperties,
     connection_status::{ConnectionState, ConnectionStatus, ConnectionStep},
+    experimental::RecoveryConfig,
     frames::Frames,
     heartbeat::Heartbeat,
     internal_rpc::{InternalRPC, InternalRPCHandle},
@@ -53,6 +54,7 @@ impl Connection {
         internal_rpc: InternalRPCHandle,
         frames: Frames,
         executor: Arc<dyn FullExecutor + Send + Sync>,
+        recovery_config: RecoveryConfig,
     ) -> Self {
         let configuration = Configuration::default();
         let status = ConnectionStatus::default();
@@ -65,6 +67,7 @@ impl Connection {
             internal_rpc.clone(),
             frames,
             executor,
+            recovery_config,
         );
         let closer = Arc::new(ConnectionCloser::new(status.clone(), internal_rpc));
         let connection = Self {
@@ -343,6 +346,7 @@ impl Connection {
             internal_rpc.handle(),
             frames.clone(),
             executor.clone(),
+            options.recovery_config.clone().unwrap_or_default(),
         );
         let status = conn.status.clone();
         let configuration = conn.configuration.clone();
@@ -507,6 +511,7 @@ mod tests {
             internal_rpc.handle(),
             Frames::default(),
             executor.clone(),
+            RecoveryConfig::default(),
         );
         conn.status.set_state(ConnectionState::Connected);
         conn.configuration.set_channel_max(2047);
@@ -586,6 +591,7 @@ mod tests {
             internal_rpc.handle(),
             Frames::default(),
             executor.clone(),
+            RecoveryConfig::default(),
         );
         conn.status.set_state(ConnectionState::Connected);
         conn.configuration.set_channel_max(2047);