fix: pass the whole buffer for reading the response

Author: rustybee42

shared/src/conn/outgoing.rs

@@ -3,6 +3,7 @@ use super::store::Store;
 use crate::bee_msg::misc::AuthenticateChannel;
 use crate::bee_msg::{Header, Msg, deserialize_body, deserialize_header, serialize};
 use crate::bee_serde::{Deserializable, Serializable};
+use crate::conn::TCP_BUF_LEN;
 use crate::conn::store::StoredStream;
 use crate::conn::stream::Stream;
 use crate::types::{AuthSecret, Uid};
@@ -52,9 +53,7 @@ impl Pool {
         let mut buf = self.store.pop_buf_or_create();
 
         let msg_len = serialize(msg, &mut buf)?;
-        let resp_header = self
-            .comm_stream(node_uid, &mut buf[0..msg_len], true)
-            .await?;
+        let resp_header = self.comm_stream(node_uid, &mut buf, msg_len, true).await?;
         let resp_msg = deserialize_body(&resp_header, &buf[Header::LEN..])?;
 
         self.store.push_buf(buf);
@@ -71,8 +70,7 @@ impl Pool {
         let mut buf = self.store.pop_buf_or_create();
 
         let msg_len = serialize(msg, &mut buf)?;
-        self.comm_stream(node_uid, &mut buf[0..msg_len], false)
-            .await?;
+        self.comm_stream(node_uid, &mut buf, msg_len, false).await?;
 
         self.store.push_buf(buf);
 
@@ -95,12 +93,15 @@ impl Pool {
         &self,
         node_uid: Uid,
         buf: &mut [u8],
+        send_len: usize,
         expect_response: bool,
     ) -> Result<Header> {
+        debug_assert_eq!(buf.len(), TCP_BUF_LEN);
+
         // 1. Pop open streams until communication succeeds or none are left
         while let Some(stream) = self.store.try_pop_stream(node_uid) {
             match self
-                .write_and_read_stream(buf, stream, expect_response)
+                .write_and_read_stream(buf, stream, send_len, expect_response)
                 .await
             {
                 Ok(header) => return Ok(header),
@@ -147,7 +148,7 @@ impl Pool {
                         // Communication using the newly opened stream should usually not fail. If
                         // it does, abort. It might be better to just try the next address though.
                         let resp_header = self
-                            .write_and_read_stream(buf, stream, expect_response)
+                            .write_and_read_stream(buf, stream, send_len, expect_response)
                             .await
                             .with_context(err_context)?;
 
@@ -166,7 +167,7 @@ impl Pool {
         let stream = self.store.pop_stream(node_uid).await?;
 
         let resp_header = self
-            .write_and_read_stream(buf, stream, expect_response)
+            .write_and_read_stream(buf, stream, send_len, expect_response)
             .await
             .with_context(|| {
                 format!("Communication using existing stream to {node_uid:?} failed")
@@ -181,9 +182,10 @@ impl Pool {
         &self,
         buf: &mut [u8],
         mut stream: StoredStream,
+        send_len: usize,
         expect_response: bool,
     ) -> Result<Header> {
-        stream.as_mut().write_all(buf).await?;
+        stream.as_mut().write_all(&buf[0..send_len]).await?;
 
         let header = if expect_response {
             // Read header