Merge branch 'rabbitmq:main' into main

Author: grahambinns

protocol/src/response/mod.rs

@@ -43,7 +43,7 @@ pub enum ResponseCode {
     FrameTooLarge,
     InternalError,
     AccessRefused,
-    PrecoditionFailed,
+    PreconditionFailed,
     PublisherDoesNotExist,
     OffsetNotFound,
 }

protocol/src/response/shims.rs

@@ -32,7 +32,7 @@ impl TryFrom<u16> for ResponseCode {
             RESPONSE_CODE_FRAME_TOO_LARGE => Ok(ResponseCode::FrameTooLarge),
             RESPONSE_CODE_INTERNAL_ERROR => Ok(ResponseCode::InternalError),
             RESPONSE_CODE_ACCESS_REFUSED => Ok(ResponseCode::AccessRefused),
-            RESPONSE_CODE_PRECONDITION_FAILED => Ok(ResponseCode::PrecoditionFailed),
+            RESPONSE_CODE_PRECONDITION_FAILED => Ok(ResponseCode::PreconditionFailed),
             RESPONSE_CODE_PUBLISHER_DOES_NOT_EXIST => Ok(ResponseCode::PublisherDoesNotExist),
             RESPONSE_CODE_OFFSET_NOT_FOUND => Ok(ResponseCode::OffsetNotFound),
             _ => Err(DecodeError::UnknownResponseCode(value)),
@@ -65,7 +65,7 @@ impl From<&ResponseCode> for u16 {
             ResponseCode::FrameTooLarge => RESPONSE_CODE_FRAME_TOO_LARGE,
             ResponseCode::InternalError => RESPONSE_CODE_INTERNAL_ERROR,
             ResponseCode::AccessRefused => RESPONSE_CODE_ACCESS_REFUSED,
-            ResponseCode::PrecoditionFailed => RESPONSE_CODE_PRECONDITION_FAILED,
+            ResponseCode::PreconditionFailed => RESPONSE_CODE_PRECONDITION_FAILED,
             ResponseCode::PublisherDoesNotExist => RESPONSE_CODE_PUBLISHER_DOES_NOT_EXIST,
             ResponseCode::OffsetNotFound => RESPONSE_CODE_OFFSET_NOT_FOUND,
         }

src/bin/perf-producer.rs

@@ -78,14 +78,14 @@ async fn main() {
             metrics = Vec::with_capacity(50 * 60 * minutes as usize);
             tokio::spawn(async move {
                 let stats = calculate_stats(last_metrics).await;
-                println!("stats: {:?}", stats);
+                println!("stats: {stats:?}");
             });
         }
         metrics.push(metric);
     }
 
     let stats = calculate_stats(metrics).await;
-    println!("stats: {:?}", stats);
+    println!("stats: {stats:?}");
 }
 
 async fn calculate_stats(metrics: Vec<Metric>) -> Stats {

src/consumer.rs

@@ -160,7 +160,7 @@ impl ConsumerBuilder {
                 return Err(ConsumerCreateError::FilteringNotSupport);
             }
             for (index, item) in filter_input.filter_values.iter().enumerate() {
-                let key = format!("filter.{}", index);
+                let key = format!("filter.{index}");
                 self.properties.insert(key, item.to_owned());
             }
 

src/stream_creator.rs

@@ -41,19 +41,22 @@ impl StreamCreator {
         binding_keys: Option<Vec<String>>,
     ) -> Result<(), StreamCreateError> {
         let mut partitions_names = Vec::with_capacity(number_of_partitions);
-        let mut new_binding_keys: Vec<String> = Vec::with_capacity(number_of_partitions);
-
-        if binding_keys.is_none() {
-            for i in 0..number_of_partitions {
-                new_binding_keys.push(i.to_string());
-                partitions_names.push(super_stream.to_owned() + "-" + i.to_string().as_str())
-            }
+        let new_binding_keys: Vec<String> = if let Some(keys) = binding_keys {
+            // Use the provided binding keys
+            keys.iter()
+                .map(|binding_key| {
+                    partitions_names.push(super_stream.to_owned() + "-" + binding_key);
+                    binding_key.clone()
+                })
+                .collect()
         } else {
-            new_binding_keys = binding_keys.unwrap();
-            for binding_key in new_binding_keys.iter() {
-                partitions_names.push(super_stream.to_owned() + "-" + binding_key)
-            }
-        }
+            (0..number_of_partitions)
+                .map(|i| {
+                    partitions_names.push(super_stream.to_owned() + "-" + &i.to_string());
+                    i.to_string()
+                })
+                .collect()
+        };
 
         let client = self.env.create_client().await?;
         let response = client

tests/client_test.rs

@@ -295,7 +295,7 @@ async fn client_declare_delete_publisher_twice_error() {
         .await
         .unwrap();
 
-    assert_eq!(&ResponseCode::PrecoditionFailed, response_error.code());
+    assert_eq!(&ResponseCode::PreconditionFailed, response_error.code());
 
     let response = test.client.delete_publisher(1).await.unwrap();
     assert_eq!(&ResponseCode::Ok, response.code());