fix clippy issue

Author: DanielePalaia

src/consumer.rs

@@ -251,7 +251,7 @@ impl Consumer {
 impl Stream for Consumer {
     type Item = Result<Delivery, ConsumerDeliveryError>;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         self.internal.waker.register(cx.waker());
         let poll = Pin::new(&mut self.receiver.lock().unwrap()).poll_recv(cx);
         match (self.is_closed(), poll.is_ready()) {

src/environment.rs

@@ -60,7 +60,7 @@ impl Environment {
             environment: self.clone(),
             data: PhantomData,
             //filter_value_extractor: None,
-            routing_strategy: routing_strategy,
+            route_strategy: routing_strategy,
         }
     }
 

src/superstream.rs

@@ -14,16 +14,16 @@ pub struct DefaultSuperStreamMetadata {
 
 impl DefaultSuperStreamMetadata {
     pub async fn partitions(&mut self) -> Vec<String> {
-        if self.partitions.len() == 0 {
+        if self.partitions.is_empty() {
             println!("partition len is 0");
             let response = self.client.partitions(self.super_stream.clone()).await;
 
             self.partitions = response.unwrap().streams;
         }
-        return self.partitions.clone();
+        self.partitions.clone()
     }
     pub async fn routes(&mut self, routing_key: String) -> Vec<String> {
-        if self.routes.len() == 0 {
+        if self.routes.is_empty() {
             let response = self
                 .client
                 .route(routing_key, self.super_stream.clone())
@@ -32,7 +32,7 @@ impl DefaultSuperStreamMetadata {
             self.routes = response.unwrap().streams;
         }
 
-        return self.routes.clone();
+        self.routes.clone()
     }
 }
 
@@ -49,9 +49,7 @@ impl RoutingKeyRoutingStrategy {
     ) -> Vec<String> {
         let key = (self.routing_extractor)(message);
 
-        let routes = metadata.routes(key).await;
-
-        return routes;
+        metadata.routes(key).await
     }
 }
 
@@ -78,7 +76,7 @@ impl HashRoutingMurmurStrategy {
         let stream = partitions.into_iter().nth(route as usize).unwrap();
         streams.push(stream);
 
-        return streams;
+        streams
     }
 }
 

src/superstream_consumer.rs

@@ -1,14 +1,7 @@
-use futures::Stream;
-use rabbitmq_stream_protocol::commands::subscribe::OffsetSpecification;
-use std::pin::Pin;
-use std::sync::Arc;
-use std::sync::Mutex;
-use std::task::{Context, Poll};
-
-use crate::consumer::Delivery;
-use crate::error::ConsumerDeliveryError;
 use crate::superstream::DefaultSuperStreamMetadata;
 use crate::{error::ConsumerCreateError, Client, Consumer, Environment};
+use rabbitmq_stream_protocol::commands::subscribe::OffsetSpecification;
+use std::sync::Arc;
 //type FilterPredicate = Option<Arc<dyn Fn(&Message) -> bool + Send + Sync>>;
 
 /// API for consuming RabbitMQ stream messages

src/superstream_producer.rs

@@ -26,7 +26,7 @@ pub struct SuperStreamProducer<T>(
 pub struct SuperStreamProducerBuilder<T> {
     pub(crate) environment: Environment,
     //pub filter_value_extractor: Option<FilterValueExtractor>,
-    pub routing_strategy: RoutingStrategy,
+    pub route_strategy: RoutingStrategy,
     pub(crate) data: PhantomData<T>,
 }
 
@@ -67,7 +67,7 @@ impl SuperStreamProducer<NoDedup> {
             }
 
             let producer = self.1.get(route.as_str()).unwrap();
-            let _ = producer.send(message.clone(), cb.clone()).await?;
+            producer.send(message.clone(), cb.clone()).await?;
         }
         Ok(())
     }
@@ -79,19 +79,16 @@ impl SuperStreamProducer<NoDedup> {
         let mut is_error = false;
         for (_, producer) in self.1.into_iter() {
             let close = producer.close().await;
-            match close {
-                Err(e) => {
-                    is_error = true;
-                    err = Some(e);
-                }
-                _ => (),
+            if let Err(e) = close {
+                is_error = true;
+                err = Some(e);
             }
         }
 
-        if is_error == false {
-            return Ok(());
+        if !is_error {
+            Ok(())
         } else {
-            return Err(err.unwrap());
+            Err(err.unwrap())
         }
     }
 }
@@ -119,7 +116,7 @@ impl<T> SuperStreamProducerBuilder<T> {
             environment: self.environment.clone(),
             client,
             //filter_value_extractor: self.filter_value_extractor,
-            routing_strategy: self.routing_strategy,
+            routing_strategy: self.route_strategy,
         };
 
         let internal_producer = Arc::new(super_stream_producer);