Merge branch 'master' into feat/transactions

Author: BewareMyPower

src/connection_manager.rs

@@ -68,6 +68,59 @@ impl Default for OperationRetryOptions {
     }
 }
 
+impl OperationRetryOptions {
+    pub fn allow_retry(&self, current: u32) -> bool {
+        self.max_retries.is_none() || current < self.max_retries.unwrap()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_allow_retry_no_max_retries() {
+        let options = OperationRetryOptions {
+            operation_timeout: Duration::from_secs(30),
+            retry_delay: Duration::from_secs(5),
+            max_retries: None,
+        };
+
+        // If max_retries is None, it should always allow retries
+        assert!(options.allow_retry(0));
+        assert!(options.allow_retry(100));
+        assert!(options.allow_retry(u32::MAX));
+    }
+
+    #[test]
+    fn test_allow_retry_with_max_retries() {
+        let options = OperationRetryOptions {
+            operation_timeout: Duration::from_secs(30),
+            retry_delay: Duration::from_secs(5),
+            max_retries: Some(3),
+        };
+
+        // If max_retries is set to 3, we allow retries for current < 3
+        assert!(options.allow_retry(0)); // current < 3
+        assert!(options.allow_retry(2)); // current < 3
+        assert!(!options.allow_retry(3)); // current == 3
+        assert!(!options.allow_retry(4)); // current > 3
+    }
+
+    #[test]
+    fn test_allow_retry_max_retries_is_zero() {
+        let options = OperationRetryOptions {
+            operation_timeout: Duration::from_secs(30),
+            retry_delay: Duration::from_secs(5),
+            max_retries: Some(0),
+        };
+
+        // If max_retries is 0, it should not allow any retries
+        assert!(!options.allow_retry(0)); // current == 0
+        assert!(!options.allow_retry(1)); // current > 0
+    }
+}
+
 /// configuration for TLS connections
 #[derive(Debug, Clone)]
 pub struct TlsOptions {

src/producer.rs

@@ -1194,7 +1194,7 @@ impl<'a, T, Exe: Executor> MessageBuilder<'a, T, Exe> {
     }
 }
 
-impl<'a, T: SerializeMessage + Sized, Exe: Executor> MessageBuilder<'a, T, Exe> {
+impl<T: SerializeMessage + Sized, Exe: Executor> MessageBuilder<'_, T, Exe> {
     /// sends the message through the producer that created it
     #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
     #[deprecated = "instead use send_non_blocking"]

src/retry_op.rs

@@ -54,26 +54,23 @@ pub async fn handle_retry_error<Exe: Executor>(
             return Err(err.into());
         }
     };
-    match operation_retry_options.max_retries {
-        Some(max_retries) if current_retries < max_retries => {
-            error!(
-                "{operation_name}({topic}) answered {kind}{text}, retrying request after {:?} (max_retries = {max_retries})",
-                operation_retry_options.retry_delay
-            );
-            client
-                .executor
-                .delay(operation_retry_options.retry_delay)
-                .await;
-
-            *addr = client.lookup_topic(topic).await?;
-            *connection = client.manager.get_connection(addr).await?;
-            Ok(())
-        }
-        _ => {
-            error!("{operation_name}({topic}) answered {kind}{text}, reached max retries");
-            Err(err.into())
-        }
+    if !(operation_retry_options.allow_retry(current_retries)) {
+        error!("{operation_name}({topic}) answered {kind}{text}, reached max retries");
+        return Err(err.into());
     }
+    error!(
+        "{operation_name}({topic}) answered {kind}{text}, retrying request after {:?} (max_retries = {:?})",
+        operation_retry_options.retry_delay,
+        operation_retry_options.max_retries
+    );
+    client
+        .executor
+        .delay(operation_retry_options.retry_delay)
+        .await;
+
+    *addr = client.lookup_topic(topic).await?;
+    *connection = client.manager.get_connection(addr).await?;
+    Ok(())
 }
 
 pub async fn retry_subscribe_consumer<Exe: Executor>(