some refactoring

Author: DanielePalaia

protocol/src/commands/superstream_partitions.rs

@@ -2,10 +2,7 @@ use std::io::Write;
 
 #[cfg(test)]
 use fake::Fake;
-<<<<<<< HEAD
-=======
 
->>>>>>> 4a7cb45 (refactoring tests)
 use super::Command;
 use crate::{
     codec::{Decoder, Encoder},

src/client/mod.rs

@@ -291,14 +291,14 @@ impl Client {
                 properties,
             )
         })
-            .await
+        .await
     }
 
     pub async fn unsubscribe(&self, subscription_id: u8) -> RabbitMQStreamResult<GenericResponse> {
         self.send_and_receive(|correlation_id| {
             UnSubscribeCommand::new(correlation_id, subscription_id)
         })
-            .await
+        .await
     }
 
     pub async fn partitions(
@@ -330,7 +330,7 @@ impl Client {
         self.send_and_receive(|correlation_id| {
             CreateStreamCommand::new(correlation_id, stream.to_owned(), options)
         })
-            .await
+        .await
     }
 
     pub async fn create_super_stream(
@@ -349,7 +349,7 @@ impl Client {
                 options,
             )
         })
-            .await
+        .await
     }
 
     pub async fn delete_stream(&self, stream: &str) -> RabbitMQStreamResult<GenericResponse> {
@@ -364,7 +364,7 @@ impl Client {
         self.send_and_receive(|correlation_id| {
             DeleteSuperStreamCommand::new(correlation_id, super_stream.to_owned())
         })
-            .await
+        .await
     }
 
     pub async fn credit(&self, subscription_id: u8, credit: u16) -> RabbitMQStreamResult<()> {
@@ -391,7 +391,7 @@ impl Client {
             stream.to_owned(),
             offset,
         ))
-            .await
+        .await
     }
 
     pub async fn query_offset(&self, reference: String, stream: &str) -> Result<u64, ClientError> {
@@ -422,7 +422,7 @@ impl Client {
                 stream.to_owned(),
             )
         })
-            .await
+        .await
     }
 
     pub async fn delete_publisher(
@@ -432,7 +432,7 @@ impl Client {
         self.send_and_receive(|correlation_id| {
             DeletePublisherCommand::new(correlation_id, publisher_id)
         })
-            .await
+        .await
     }
 
     pub async fn publish<T: BaseMessage>(
@@ -475,8 +475,8 @@ impl Client {
         self.send_and_receive::<QueryPublisherResponse, _, _>(|correlation_id| {
             QueryPublisherRequest::new(correlation_id, reference.to_owned(), stream.to_owned())
         })
-            .await
-            .map(|sequence| sequence.from_response())
+        .await
+        .map(|sequence| sequence.from_response())
     }
 
     pub async fn exchange_command_versions(
@@ -485,7 +485,7 @@ impl Client {
         self.send_and_receive::<ExchangeCommandVersionsResponse, _, _>(|correlation_id| {
             ExchangeCommandVersionsRequest::new(correlation_id, vec![])
         })
-            .await
+        .await
     }
 
     pub fn filtering_supported(&self) -> bool {
@@ -509,13 +509,13 @@ impl Client {
                 let roots = Self::build_root_store(Some(Path::new(
                     &broker.tls.get_root_certificates_path(),
                 )))
-                    .await?;
+                .await?;
                 config = Self::build_tls_client_configuration(
                     broker.tls.get_client_certificates_path(),
                     &roots,
                     broker,
                 )
-                    .await?;
+                .await?;
             } else {
                 config = Self::build_tls_client_configuration_untrusted().await?;
             }
@@ -547,15 +547,15 @@ impl Client {
         self.with_state_lock(self.peer_properties(), move |state, server_properties| {
             state.server_properties = server_properties;
         })
-            .await?;
+        .await?;
         self.authenticate().await?;
 
         self.wait_for_tune_data().await?;
 
         self.with_state_lock(self.open(), |state, connection_properties| {
             state.connection_properties = connection_properties;
         })
-            .await?;
+        .await?;
 
         // Start heartbeat task after connection is established
         self.start_hearbeat_task(self.state.write().await.deref_mut());
@@ -619,8 +619,8 @@ impl Client {
         self.send_and_receive::<SaslHandshakeResponse, _, _>(|correlation_id| {
             SaslHandshakeCommand::new(correlation_id)
         })
-            .await
-            .map(|handshake| handshake.mechanisms)
+        .await
+        .map(|handshake| handshake.mechanisms)
     }
 
     async fn send_and_receive<T, R, M>(&self, msg_factory: M) -> Result<T, ClientError>
@@ -664,22 +664,22 @@ impl Client {
         self.send_and_receive::<OpenResponse, _, _>(|correlation_id| {
             OpenCommand::new(correlation_id, self.opts.v_host.clone())
         })
-            .await
-            .and_then(|open| {
-                if open.is_ok() {
-                    Ok(open.connection_properties)
-                } else {
-                    Err(ClientError::RequestError(open.code().clone()))
-                }
-            })
+        .await
+        .and_then(|open| {
+            if open.is_ok() {
+                Ok(open.connection_properties)
+            } else {
+                Err(ClientError::RequestError(open.code().clone()))
+            }
+        })
     }
 
     async fn peer_properties(&self) -> Result<HashMap<String, String>, ClientError> {
         self.send_and_receive::<PeerPropertiesResponse, _, _>(|correlation_id| {
             PeerPropertiesCommand::new(correlation_id, HashMap::new())
         })
-            .await
-            .map(|peer_properties| peer_properties.server_properties)
+        .await
+        .map(|peer_properties| peer_properties.server_properties)
     }
 
     async fn handle_tune_command(&self, tunes: &TunesCommand) {
@@ -723,7 +723,7 @@ impl Client {
                 tokio::time::sleep(Duration::from_secs(heartbeat_interval.into())).await;
             }
         })
-            .into();
+        .into();
         state.heartbeat_task = Some(heartbeat_task);
     }
 
@@ -781,7 +781,7 @@ impl Client {
             let client_certs = Self::build_client_certificates(Path::new(
                 &broker.tls.get_client_certificates_path(),
             ))
-                .await?;
+            .await?;
             let client_keys =
                 Self::build_client_private_keys(Path::new(&broker.tls.get_client_keys_path()))
                     .await?;
@@ -826,4 +826,4 @@ impl Client {
 
         Ok(config)
     }
-}
+}

src/environment.rs

@@ -12,8 +12,8 @@ use crate::{
     producer::ProducerBuilder,
     stream_creator::StreamCreator,
     superstream::RoutingStrategy,
-    superstream_producer::SuperStreamProducerBuilder,
     superstream_consumer::SuperStreamConsumerBuilder,
+    superstream_producer::SuperStreamProducerBuilder,
     RabbitMQStreamResult,
 };
 

src/lib.rs

@@ -99,8 +99,8 @@ pub mod types {
     pub use crate::stream_creator::StreamCreator;
     pub use crate::superstream::HashRoutingMurmurStrategy;
     pub use crate::superstream::RoutingKeyRoutingStrategy;
-    pub use crate::superstream_consumer::SuperStreamConsumer;
     pub use crate::superstream::RoutingStrategy;
+    pub use crate::superstream_consumer::SuperStreamConsumer;
     pub use rabbitmq_stream_protocol::message::Message;
     pub use rabbitmq_stream_protocol::{Response, ResponseCode, ResponseKind};
 

src/superstream_consumer.rs

@@ -1,23 +1,23 @@
 use std::sync::Arc;
 
-use rabbitmq_stream_protocol::{commands::subscribe::OffsetSpecification, message::Message};
+use rabbitmq_stream_protocol::commands::subscribe::OffsetSpecification;
 
 use crate::superstream::DefaultSuperStreamMetadata;
 use crate::{error::ConsumerCreateError, Client, Consumer, Environment};
 
-type FilterPredicate = Option<Arc<dyn Fn(&Message) -> bool + Send + Sync>>;
+//type FilterPredicate = Option<Arc<dyn Fn(&Message) -> bool + Send + Sync>>;
 
 /// API for consuming RabbitMQ stream messages
 #[derive(Clone)]
 pub struct SuperStreamConsumer {
-    pub internal: Arc<SuperStreamConsumerInternal>,
+    internal: Arc<SuperStreamConsumerInternal>,
 }
 
 struct SuperStreamConsumerInternal {
     client: Client,
     super_stream: String,
     offset_specification: OffsetSpecification,
-    pub consumers: Vec<Consumer>,
+    consumers: Vec<Consumer>,
 }
 
 /// Builder for [`Consumer`]
@@ -68,15 +68,14 @@ impl SuperStreamConsumerBuilder {
         })
     }
 
-    pub fn offset(mut self, offset_specification: OffsetSpecification) -> Self {
+    pub async fn offset(mut self, offset_specification: OffsetSpecification) -> Self {
         self.offset_specification = offset_specification;
         self
     }
 }
 
-impl SuperStreamConsumer   {
-
-    pub async fn get_consumers(&self) -> &Vec<Consumer>   {
-        return &self.internal.consumers
+impl SuperStreamConsumer {
+    pub async fn get_consumers(&self) -> &Vec<Consumer> {
+        return &self.internal.consumers;
     }
 }

tests/integration/client_test.rs

@@ -447,4 +447,3 @@ async fn client_test_route_test() {
         test.partitions.get(0).unwrap()
     );
 }
-

tests/integration/consumer_test.rs

@@ -3,7 +3,6 @@ use std::time::Duration;
 use crate::common::TestEnvironment;
 use fake::{Fake, Faker};
 use futures::StreamExt;
-use tokio::task;
 use rabbitmq_stream_client::{
     error::{
         ClientError, ConsumerCloseError, ConsumerDeliveryError, ConsumerStoreOffsetError,
@@ -12,11 +11,12 @@ use rabbitmq_stream_client::{
     types::{Delivery, Message, OffsetSpecification, SuperStreamConsumer},
     Consumer, FilterConfiguration, NoDedup, Producer,
 };
+use tokio::task;
 
+use rabbitmq_stream_client::types::{HashRoutingMurmurStrategy, RoutingStrategy};
 use rabbitmq_stream_protocol::ResponseCode;
-use std::sync::Arc;
 use std::sync::atomic::{AtomicU32, Ordering};
-use rabbitmq_stream_client::types::{HashRoutingMurmurStrategy, RoutingStrategy};
+use std::sync::Arc;
 
 #[tokio::test(flavor = "multi_thread")]
 async fn consumer_test() {
@@ -79,13 +79,14 @@ async fn super_stream_consumer_test() {
         .await
         .unwrap();
 
-    static super_stream_consumer: SuperStreamConsumer = env
-        .env
-        .super_stream_consumer()
-        .offset(OffsetSpecification::Next)
-        .build(&env.stream)
-        .await
-        .unwrap();
+    let super_stream_consumer: Arc<SuperStreamConsumer> = Arc::new(
+        env.env
+            .super_stream_consumer()
+            //.offset(OffsetSpecification::Next)
+            .build(&env.stream)
+            .await
+            .unwrap(),
+    );
 
     for n in 0..message_count {
         let msg = Message::builder().body(format!("message{}", n)).build();
@@ -97,22 +98,19 @@ async fn super_stream_consumer_test() {
             .unwrap();
     }
 
-
     let received_messages = Arc::new(AtomicU32::new(0));
-    let consumers =  super_stream_consumer.get_consumers().await;
 
     let mut tasks = Vec::new();
-        for mut consumer in consumers.into_iter() {
-            let received_messages_outer = received_messages.clone();
-            tasks.push(task::spawn(async move {
-                let inner_received_messages = received_messages_outer.clone();
-                let delivery = consumer.next().await.unwrap();
-                let _ = String::from_utf8(delivery.unwrap().message().data().unwrap().to_vec()).unwrap();
-                inner_received_messages.fetch_add(1, Ordering::Relaxed);
-
-            }));
-        }
-
+    for mut consumer in super_stream_consumer.get_consumers().await.into_iter() {
+        let received_messages_outer = received_messages.clone();
+        tasks.push(task::spawn(async move {
+            let inner_received_messages = received_messages_outer.clone();
+            let delivery = consumer.next().await.unwrap();
+            let _ =
+                String::from_utf8(delivery.unwrap().message().data().unwrap().to_vec()).unwrap();
+            inner_received_messages.fetch_add(1, Ordering::Relaxed);
+        }));
+    }
     futures::future::join_all(tasks).await;
 
     assert!(received_messages.fetch_add(1, Ordering::Relaxed) == message_count);
@@ -398,7 +396,7 @@ async fn consumer_test_with_store_offset() {
         // Store an offset
         if i == offset_to_store {
             //Store the 5th element produced
-            let result = consumer_store
+            let _ = consumer_store
                 .store_offset(delivery.unwrap().offset())
                 .await;
         }