Address part of PR comments

common/client-core/src/client/inbound_messages.rs

@@ -6,6 +6,7 @@ use nym_sphinx::forwarding::packet::MixPacket;
 use nym_sphinx::params::PacketType;
 use nym_task::connections::TransmissionLane;
 use serde::{Deserialize, Serialize};
+use std::convert::TryInto;
 use tokio_util::{
     bytes::Buf,
     bytes::BytesMut,
@@ -17,6 +18,8 @@ use crate::error::ClientCoreError;
 pub type InputMessageSender = tokio_util::sync::PollSender<InputMessage>;
 pub type InputMessageReceiver = tokio::sync::mpsc::Receiver<InputMessage>;
 
+const LENGHT_ENCODING_PREFIX_SIZE: usize = 4;
+
 #[derive(Serialize, Deserialize, Debug)]
 pub enum InputMessage {
     /// Fire an already prepared mix packets into the network.
@@ -198,11 +201,25 @@ impl InputMessage {
     }
 
     pub fn serialized_size(&self) -> u64 {
-        bincode::serialized_size(self).expect("failed to get serialized InputMessage size") + 4
+        bincode::serialized_size(self).expect("failed to get serialized InputMessage size")
+            + LENGHT_ENCODING_PREFIX_SIZE as u64
     }
 }
 
 // TODO: Tests
+pub struct AdressedInputMessageCodec(pub Recipient);
+
+impl Encoder<&[u8]> for AdressedInputMessageCodec {
+    type Error = ClientCoreError;
+
+    fn encode(&mut self, item: &[u8], buf: &mut BytesMut) -> Result<(), Self::Error> {
+        let mut codec = InputMessageCodec;
+        let input_message = InputMessage::simple(item, self.0.clone());
+        codec.encode(input_message, buf)?;
+        Ok(())
+    }
+}
+
 pub struct InputMessageCodec;
 
 impl Encoder<InputMessage> for InputMessageCodec {
@@ -224,21 +241,27 @@ impl Decoder for InputMessageCodec {
     type Error = ClientCoreError;
 
     fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        if buf.len() < 4 {
+        if buf.len() < LENGHT_ENCODING_PREFIX_SIZE {
             return Ok(None);
         }
 
-        let len = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;
-        if buf.len() < len + 4 {
+        let len = u32::from_le_bytes(
+            buf[0..LENGHT_ENCODING_PREFIX_SIZE]
+                .try_into()
+                .expect("Could not coarce to array"),
+        ) as usize;
+        if buf.len() < len + LENGHT_ENCODING_PREFIX_SIZE {
             return Ok(None);
         }
 
-        let decoded = match bincode::deserialize(&buf[4..len + 4]) {
+        let decoded = match bincode::deserialize(
+            &buf[LENGHT_ENCODING_PREFIX_SIZE..len + LENGHT_ENCODING_PREFIX_SIZE],
+        ) {
             Ok(decoded) => decoded,
             Err(_) => return Ok(None),
         };
 
-        buf.advance(len + 4);
+        buf.advance(len + LENGHT_ENCODING_PREFIX_SIZE);
 
         Ok(Some(decoded))
     }

common/nymsphinx/src/receiver.rs

@@ -63,7 +63,7 @@ impl From<PlainMessage> for ReconstructedMessage {
 }
 
 pub struct ReconstructedMessageCodec;
-const OFFSET: usize = 4;
+const LENGHT_ENCODING_PREFIX_SIZE: usize = 4;
 
 impl Encoder<ReconstructedMessage> for ReconstructedMessageCodec {
     type Error = MessageRecoveryError;
@@ -88,21 +88,23 @@ impl Decoder for ReconstructedMessageCodec {
     type Error = MessageRecoveryError;
 
     fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        if buf.len() < OFFSET {
+        if buf.len() < LENGHT_ENCODING_PREFIX_SIZE {
             return Ok(None);
         }
 
         let len = u32::from_le_bytes(
-            buf[0..OFFSET]
+            buf[0..LENGHT_ENCODING_PREFIX_SIZE]
                 .try_into()
-                .expect("We know that we have at least OFFSET bytes in there"),
+                .expect("We know that we have at least LENGHT_ENCODING_PREFIX_SIZE bytes in there"),
         ) as usize;
 
-        if buf.len() < len + OFFSET {
+        if buf.len() < len + LENGHT_ENCODING_PREFIX_SIZE {
             return Ok(None);
         }
 
-        let decoded = match bincode::deserialize(&buf[OFFSET..len + OFFSET]) {
+        let decoded = match bincode::deserialize(
+            &buf[LENGHT_ENCODING_PREFIX_SIZE..len + LENGHT_ENCODING_PREFIX_SIZE],
+        ) {
             Ok(decoded) => decoded,
             Err(e) => {
                 debug!("Failed to decode the message - {:?}", e);
@@ -110,7 +112,7 @@ impl Decoder for ReconstructedMessageCodec {
             }
         };
 
-        buf.advance(len + 4);
+        buf.advance(len + LENGHT_ENCODING_PREFIX_SIZE);
 
         Ok(Some(decoded))
     }

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

@@ -384,7 +384,7 @@ impl MixnetClient {
             Poll::Ready(Ok(()))
         } else {
             let written = buf.capacity();
-            buf.put_slice(&self._read.buffer[..written]);
+            buf.put_slice(&self._read.buffer.split_off(written));
             self._read.buffer.advance(written);
             cx.waker().wake_by_ref();
             Poll::Ready(Ok(()))
@@ -426,11 +426,16 @@ impl AsyncRead for MixnetClient {
             Poll::Pending => return Poll::Pending,
         };
 
-        // let mut buffer = BytesMut::new();
-
-        codec.encode(msg, &mut self._read.buffer).unwrap();
-
-        // = buffer.to_vec();
+        match codec.encode(msg, &mut self._read.buffer) {
+            Ok(_) => {}
+            Err(e) => {
+                error!("failed to encode reconstructed message: {:?}", e);
+                return Poll::Ready(Err(tokio::io::Error::new(
+                    tokio::io::ErrorKind::Other,
+                    "failed to encode reconstructed message",
+                )));
+            }
+        };
 
         self.read_buffer_to_slice(buf, cx)
     }