AsyncRead for MixnetClient

common/nymsphinx/anonymous-replies/src/requests/mod.rs

@@ -7,7 +7,6 @@ use crate::{ReplySurbError, ReplySurbWithKeyRotation};
 use nym_sphinx_addressing::clients::{Recipient, RecipientFormattingError};
 use nym_sphinx_params::key_rotation::InvalidSphinxKeyRotation;
 use rand::{CryptoRng, RngCore};
-use serde::{Deserialize, Serialize};
 use std::fmt::{Display, Formatter};
 use std::mem;
 use thiserror::Error;
@@ -31,7 +30,7 @@ pub enum InvalidAnonymousSenderTagRepresentation {
     InvalidLength { received: usize, expected: usize },
 }
 
-#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash, Serialize, Deserialize)]
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
 #[cfg_attr(target_arch = "wasm32", wasm_bindgen)]
 pub struct AnonymousSenderTag([u8; SENDER_TAG_SIZE]);
 

common/nymsphinx/src/receiver.rs

@@ -2,7 +2,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::message::{NymMessage, NymMessageError, PaddedMessage, PlainMessage};
-use log::warn;
 use nym_crypto::aes::cipher::{KeyIvInit, StreamCipher};
 use nym_crypto::asymmetric::x25519;
 use nym_crypto::shared_key::recompute_shared_key;
@@ -16,11 +15,10 @@ use nym_sphinx_chunking::reconstruction::MessageReconstructor;
 use nym_sphinx_params::{
     PacketEncryptionAlgorithm, PacketHkdfAlgorithm, ReplySurbEncryptionAlgorithm,
 };
-use serde::{Deserialize, Serialize};
 use thiserror::Error;
 
 // TODO: should this live in this file?
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug)]
 pub struct ReconstructedMessage {
     /// The actual plaintext message that was received.
     pub message: Vec<u8>,
@@ -30,30 +28,6 @@ pub struct ReconstructedMessage {
     pub sender_tag: Option<AnonymousSenderTag>,
 }
 
-impl From<ReconstructedMessage> for Vec<u8> {
-    fn from(msg: ReconstructedMessage) -> Vec<u8> {
-        match bincode::serialize(&msg) {
-            Ok(serialized) => serialized,
-            Err(err) => {
-                warn!("failed to serialize reconstructed message - {:?}", err);
-                Vec::new()
-            }
-        }
-    }
-}
-
-impl From<&ReconstructedMessage> for Vec<u8> {
-    fn from(msg: &ReconstructedMessage) -> Vec<u8> {
-        match bincode::serialize(msg) {
-            Ok(serialized) => serialized,
-            Err(err) => {
-                warn!("failed to serialize reconstructed message - {:?}", err);
-                Vec::new()
-            }
-        }
-    }
-}
-
 impl From<ReconstructedMessage> for (Vec<u8>, Option<AnonymousSenderTag>) {
     fn from(msg: ReconstructedMessage) -> Self {
         (msg.message, msg.sender_tag)

sdk/rust/nym-sdk/src/mixnet/native_client.rs

@@ -4,7 +4,7 @@ use crate::mixnet::stream::{MixnetListener, MixnetStream};
 use crate::mixnet::traits::MixnetMessageSender;
 use crate::{Error, Result};
 use async_trait::async_trait;
-use futures::{ready, Stream, StreamExt};
+use futures::{ready, AsyncRead, Stream, StreamExt};
 use log::{debug, error};
 use nym_client_core::client::base_client::GatewayConnection;
 use nym_client_core::client::mix_traffic::ClientRequestSender;
@@ -76,6 +76,24 @@ pub struct MixnetClient {
 
     /// How long a stream can be idle before the router cleans it up.
     pub(crate) stream_idle_timeout: Duration,
+
+    // internal state used for the `AsyncRead` implementation
+    _read: ReadBuffer,
+}
+
+#[derive(Debug, Default)]
+struct ReadBuffer {
+    buffer: Vec<u8>,
+}
+
+impl ReadBuffer {
+    fn clear(&mut self) {
+        self.buffer.clear();
+    }
+
+    fn pending(&self) -> bool {
+        !self.buffer.is_empty()
+    }
 }
 
 impl MixnetClient {
@@ -109,6 +127,7 @@ impl MixnetClient {
             stream_mode: Arc::new(AtomicBool::new(false)),
             streams: None,
             stream_idle_timeout: super::stream::DEFAULT_STREAM_IDLE_TIMEOUT,
+            _read: ReadBuffer::default(),
         }
     }
 
@@ -329,6 +348,25 @@ impl MixnetClient {
     pub fn listener(&mut self) -> Result<MixnetListener> {
         super::stream::listener(self)
     }
+
+    fn read_buffer_to_slice(
+        &mut self,
+        buf: &mut [u8],
+        cx: &mut Context<'_>,
+    ) -> Poll<std::result::Result<usize, std::io::Error>> {
+        if self._read.buffer.len() < buf.len() {
+            let written = self._read.buffer.len();
+            buf[..written].copy_from_slice(&self._read.buffer);
+            self._read.clear();
+            Poll::Ready(Ok(written))
+        } else {
+            let written = buf.len();
+            buf.copy_from_slice(&self._read.buffer[..written]);
+            self._read.buffer = self._read.buffer[written..].to_vec();
+            cx.waker().wake_by_ref();
+            Poll::Ready(Ok(written))
+        }
+    }
 }
 
 pub struct MixnetClientSender {
@@ -347,6 +385,28 @@ impl Clone for MixnetClientSender {
     }
 }
 
+impl AsyncRead for MixnetClient {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<std::result::Result<usize, std::io::Error>> {
+        if self._read.pending() {
+            return self.read_buffer_to_slice(buf, cx);
+        }
+
+        let msg = match self.as_mut().poll_next(cx) {
+            Poll::Ready(Some(msg)) => msg,
+            Poll::Ready(None) => return Poll::Ready(Ok(0)),
+            Poll::Pending => return Poll::Pending,
+        };
+
+        self._read.buffer = msg.message;
+
+        self.read_buffer_to_slice(buf, cx)
+    }
+}
+
 impl Stream for MixnetClient {
     type Item = ReconstructedMessage;