read_buf: use unsafe but remove callback

crates/librqbit/src/peer_connection.rs

@@ -347,23 +347,21 @@ impl<H: PeerConnectionHandler> PeerConnection<H> {
 
         let reader = async move {
             loop {
-                read_buf
+                let message = read_buf
-                    .read_message(&mut read_half, rwtimeout, |message| {
+                    .read_message(&mut read_half, rwtimeout)
-                        trace!("received: {:?}", &message);
-
-                        if let Message::Extended(ExtendedMessage::Handshake(h)) = &message {
-                            *extended_handshake_ref.write() = Some(h.clone_to_owned());
-                            self.handler.on_extended_handshake(h)?;
-                            trace!("remembered extended handshake for future serializing");
-                        } else {
-                            self.handler
-                                .on_received_message(message)
-                                .context("error in handler.on_received_message()")?;
-                        }
-                        Ok(())
-                    })
                     .await
                     .context("error reading message")?;
+                trace!("received: {:?}", &message);
+
+                if let Message::Extended(ExtendedMessage::Handshake(h)) = &message {
+                    *extended_handshake_ref.write() = Some(h.clone_to_owned());
+                    self.handler.on_extended_handshake(h)?;
+                    trace!("remembered extended handshake for future serializing");
+                } else {
+                    self.handler
+                        .on_received_message(message)
+                        .context("error in handler.on_received_message()")?;
+                }
             }
 
             // For type inference.

crates/librqbit/src/read_buf.rs

@@ -68,18 +68,17 @@ impl ReadBuf {
         &mut self,
         mut conn: impl AsyncReadExt + Unpin,
         timeout: Duration,
-        on_message: impl for<'a> FnOnce(MessageBorrowed<'a>) -> anyhow::Result<()>,
+    ) -> anyhow::Result<MessageBorrowed<'_>> {
-    ) -> anyhow::Result<()> {
         loop {
             let need_additional_bytes =
                 match MessageBorrowed::deserialize(&self.buf[self.processed..self.filled]) {
                     Err(MessageDeserializeError::NotEnoughData(d, _)) => d,
                     Ok((msg, size)) => {
                         self.processed += size;
-                        // Rust's borrow checker can't do this early return. So we are using a callback instead.
+
-                        // return Ok(msg);
+                        // Rust's borrow checker can't do this early return so resort to unsafe.
-                        on_message(msg)?;
+                        let msg = unsafe { std::mem::transmute(msg) };
-                        return Ok(());
+                        return Ok(msg);
                     }
                     Err(e) => return Err(e.into()),
                 };