Merge branch 'master' into fix/producer-small-futures

Author: BewareMyPower

src/consumer/engine.rs

@@ -110,7 +110,7 @@ impl<Exe: Executor> ConsumerEngine<Exe> {
 
             let send_end_res = event_tx.send(mapper(None)).await;
             if let Err(err) = send_end_res {
-                log::error!("Error sending end event to channel - {err}");
+                log::debug!("Error sending close event to channel - {err}");
             }
 
             log::warn!("rx terminated");

src/lib.rs

@@ -252,49 +252,6 @@ mod tests {
         }
     }
 
-    #[derive(Debug)]
-    enum Error {
-        Pulsar(PulsarError),
-        Timeout(std::io::Error),
-        Serde(serde_json::Error),
-        Utf8(std::string::FromUtf8Error),
-    }
-
-    impl From<std::io::Error> for Error {
-        fn from(e: std::io::Error) -> Self {
-            Error::Timeout(e)
-        }
-    }
-
-    impl From<PulsarError> for Error {
-        fn from(e: PulsarError) -> Self {
-            Error::Pulsar(e)
-        }
-    }
-
-    impl From<serde_json::Error> for Error {
-        fn from(e: serde_json::Error) -> Self {
-            Error::Serde(e)
-        }
-    }
-
-    impl From<std::string::FromUtf8Error> for Error {
-        fn from(err: std::string::FromUtf8Error) -> Self {
-            Error::Utf8(err)
-        }
-    }
-
-    impl std::fmt::Display for Error {
-        fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-            match self {
-                Error::Pulsar(e) => write!(f, "{e}"),
-                Error::Timeout(e) => write!(f, "{e}"),
-                Error::Serde(e) => write!(f, "{e}"),
-                Error::Utf8(e) => write!(f, "{e}"),
-            }
-        }
-    }
-
     pub struct SimpleLogger {
         pub tag: &'static str,
     }

src/message.rs

@@ -447,7 +447,7 @@ struct CommandFrame<'a> {
 }
 
 #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
-fn command_frame(i: &[u8]) -> IResult<&[u8], CommandFrame> {
+fn command_frame<'a>(i: &'a [u8]) -> IResult<&'a [u8], CommandFrame<'a>> {
     let (i, total_size) = be_u32(i)?;
     let (i, command_size) = be_u32(i)?;
     let (i, command) = take(command_size)(i)?;
@@ -473,7 +473,7 @@ struct PayloadFrame<'a> {
 }
 
 #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
-fn payload_frame(i: &[u8]) -> IResult<&[u8], PayloadFrame> {
+fn payload_frame<'a>(i: &'a [u8]) -> IResult<&'a [u8], PayloadFrame<'a>> {
     let (i, magic_number) = be_u16(i)?;
     let (i, checksum) = be_u32(i)?;
     let (i, metadata_size) = be_u32(i)?;

src/producer.rs

@@ -341,7 +341,7 @@ impl<Exe: Executor> Producer<Exe> {
     ///
     /// the created message will ber sent by this producer in [MessageBuilder::send]
     #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
-    pub fn create_message(&mut self) -> MessageBuilder<(), Exe> {
+    pub fn create_message<'a>(&'a mut self) -> MessageBuilder<'a, (), Exe> {
         MessageBuilder::new(self)
     }