chore: apply formatter

Signed-off-by: Florentin Dubois <florentin.dubois@clever-cloud.com>

src/connection.rs

@@ -655,7 +655,10 @@ impl<Exe: Executor> ConnectionSender<Exe> {
             response
                 .await
                 .map_err(|oneshot::Canceled| {
-                    error!("response has been canceled (key = {:?}), we are disconnected", k);
+                    error!(
+                        "response has been canceled (key = {:?}), we are disconnected",
+                        k
+                    );
                     error.set(ConnectionError::Disconnected);
                     ConnectionError::Disconnected
                 })
@@ -674,7 +677,10 @@ impl<Exe: Executor> ConnectionSender<Exe> {
                 let connection_id = self.connection_id;
                 let error = self.error.clone();
                 let delay_f = self.executor.delay(self.operation_timeout);
-                trace!("Create timeout futures with operation timeout at {:?}", self.operation_timeout);
+                trace!(
+                    "Create timeout futures with operation timeout at {:?}",
+                    self.operation_timeout
+                );
                 let fut = async move {
                     pin_mut!(response);
                     pin_mut!(delay_f);

src/consumer/builder.rs

@@ -177,9 +177,7 @@ impl<Exe: Executor> ConsumerBuilder<Exe> {
     // Checks the builder for inconsistencies
     // returns a config and a list of topics with associated brokers
     #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
-    async fn validate(
-        self,
-    ) -> Result<(ConsumerConfig, Vec<(String, BrokerAddress)>), Error> {
+    async fn validate(self) -> Result<(ConsumerConfig, Vec<(String, BrokerAddress)>), Error> {
         let ConsumerBuilder {
             pulsar,
             topics,

src/consumer/engine.rs

@@ -148,12 +148,17 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
                     })?;
             }
 
-            // debug!("self.remaining_messages ({}) < self.batch_size.div_ceil(2) ({}) = {}", self.remaining_messages, self.batch_size.div_ceil(2), self.remaining_messages < self.batch_size.div_ceil(2));
+            // debug!("self.remaining_messages ({}) < self.batch_size.div_ceil(2) ({}) = {}",
+            // self.remaining_messages, self.batch_size.div_ceil(2), self.remaining_messages <
+            // self.batch_size.div_ceil(2));
             if self.remaining_messages < (self.batch_size.div_ceil(2) as i64) {
                 match self
                     .connection
                     .sender()
-                    .send_flow(self.id, (self.batch_size as i64 - self.remaining_messages) as u32)
+                    .send_flow(
+                        self.id,
+                        (self.batch_size as i64 - self.remaining_messages) as u32,
+                    )
                     .await
                 {
                     Ok(()) => {}
@@ -165,20 +170,22 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
                     // channel to be not full
                     Err(e) => {
                         error!("consumer engine: we got a unrecoverable connection error, {e}");
-                        return Err(e.into())
-                    },
+                        return Err(e.into());
+                    }
                 }
 
                 self.remaining_messages = self.batch_size as i64;
             }
 
             match Self::timeout(self.event_rx.next(), Duration::from_secs(1)).await {
                 Err(_) => {
-                    // If you are reading this comment, you may have an issue where you have received a batched message
-                    // that is greater that the batch size and then break the way that we send flow command message.
+                    // If you are reading this comment, you may have an issue where you have
+                    // received a batched message that is greater that the batch
+                    // size and then break the way that we send flow command message.
                     //
-                    // In that case, you could increase your batch size or patch this driver by adding the following line,
-                    // if you are sure that you have at least 1 incoming message per second.
+                    // In that case, you could increase your batch size or patch this driver by
+                    // adding the following line, if you are sure that you have
+                    // at least 1 incoming message per second.
                     //
                     // ```rust
                     // self.remaining_messages = 0;
@@ -221,7 +228,10 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
                     .payload
                     .as_ref()
                     .and_then(|payload| {
-                        debug!("Consumer: received message payload, num_messages_in_batch = {:?}", payload.metadata.num_messages_in_batch);
+                        debug!(
+                            "Consumer: received message payload, num_messages_in_batch = {:?}",
+                            payload.metadata.num_messages_in_batch
+                        );
                         payload.metadata.num_messages_in_batch
                     })
                     .unwrap_or(1) as i64;

src/consumer/topic.rs

@@ -323,12 +323,8 @@ impl<T: DeserializeMessage, Exe: Executor> Stream for TopicConsumer<T, Exe> {
     #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         match self.messages.as_mut().poll_next(cx) {
-            Poll::Pending => {
-                Poll::Pending
-            },
-            Poll::Ready(None) => {
-                Poll::Ready(None)
-            },
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(None) => Poll::Ready(None),
             Poll::Ready(Some(Ok((id, payload)))) => {
                 self.last_message_received = Some(Utc::now());
                 self.messages_received += 1;
@@ -337,7 +333,7 @@ impl<T: DeserializeMessage, Exe: Executor> Stream for TopicConsumer<T, Exe> {
             Poll::Ready(Some(Err(e))) => {
                 error!("we are using in the single-consumer and we got an error, {e}");
                 Poll::Ready(Some(Err(e)))
-            },
+            }
         }
     }
 }

src/producer.rs

@@ -359,8 +359,8 @@ impl<Exe: Executor> Producer<Exe> {
     /// this function returns a `SendFuture` because the receipt can come long after
     /// this function was called, for various reasons:
     /// - the message was sent successfully but Pulsar did not send the receipt yet
-    /// - the producer is batching messages, so this function must return immediately,
-    ///     and the receipt will come when the batched messages are actually sent
+    /// - the producer is batching messages, so this function must return immediately, and the
+    ///   receipt will come when the batched messages are actually sent
     ///
     /// Usage:
     ///