(trying) revert changes in decode_from_file for ResponseMessage and A…

…piMessage

src/protocol/fluvio-protocol-api/src/api.rs

@@ -1,5 +1,6 @@
 use std::default::Default;
 use std::io::Error as IoError;
+use std::io::ErrorKind;
 use std::fs::File;
 use std::io::Cursor;
 use std::path::Path;
@@ -9,6 +10,7 @@ use std::fmt;
 use std::convert::TryFrom;
 
 use log::debug;
+use log::trace;
 
 use crate::core::Decoder;
 use crate::core::Encoder;
@@ -53,6 +55,17 @@ pub trait ApiMessage: Sized + Default {
         let data = buffer.to_vec();
         let mut src = Cursor::new(&data);
 
+        let mut size: i32 = 0;
+        size.decode(&mut src, 0)?;
+        trace!("decoded request size: {} bytes", size);
+
+        if src.remaining() < size as usize {
+            return Err(IoError::new(
+                ErrorKind::UnexpectedEof,
+                "not enought bytes for request message",
+            ));
+        }
+
         Self::decode_from(&mut src)
     }
 }

src/protocol/fluvio-protocol-api/src/response.rs

@@ -1,6 +1,7 @@
 use std::fs::File;
 use std::io::Cursor;
 use std::io::Error as IoError;
+use std::io::ErrorKind;
 use std::io::Read;
 use std::path::Path;
 
@@ -67,6 +68,18 @@ where
 
         let mut src = Cursor::new(&data);
 
+        // ResponseMessage implementation of fluvio_protocol::storage::FileWrite trait first encodes the length
+        // of the ResponseMessage
+        let mut size: i32 = 0;
+        size.decode(&mut src, version)?;
+        trace!("decoded response size: {} bytes", size);
+
+        if src.remaining() < size as usize {
+            return Err(IoError::new(
+                ErrorKind::UnexpectedEof,
+                "not enought for response",
+            ));
+        }
         Self::decode_from(&mut src, version)
     }
 }

src/protocol/src/store.rs

@@ -55,7 +55,8 @@ where
     }
 }
 
-/// This is same as encoding in the ResponseMessage but can encode async file slice
+/// This is same as encoding in the ResponseMessage but first
+/// includes the lenght and can encode async file slice
 impl<P> FileWrite for ResponseMessage<P>
 where
     P: FileWrite + Default,
@@ -84,6 +85,8 @@ where
     }
 }
 
+/// This is same as encoding in the RequestMessage but first
+/// includes the lenght and can encode async file slice
 impl<R> FileWrite for RequestMessage<R>
 where
     R: FileWrite + Default + Request,
@@ -94,15 +97,15 @@ where
         data: &mut Vec<StoreValue>,
         version: Version,
     ) -> Result<(), IoError> {
-        trace!("file encoding response message");
+        trace!("file encoding request message");
         let len = self.write_size(version) as i32;
-        trace!("file encoding response len: {}", len);
+        trace!("file encoding request len: {}", len);
         len.encode(dest, version)?;
 
         trace!("file encoding header");
         self.header.encode(dest, version)?;
 
-        trace!("encoding response");
+        trace!("encoding request");
         self.request.file_encode(dest, data, version)?;
         Ok(())
     }