Done!

Author: Mathieu Amiot

src/codec.rs

@@ -1,5 +1,5 @@
 use crate::error::NatsError;
-use crate::protocol::{CommandError, Op};
+use crate::protocol::Op;
 use bytes::{BufMut, BytesMut};
 use tokio_codec::{Decoder, Encoder};
 
@@ -41,14 +41,21 @@ impl Decoder for OpCodec {
             return Ok(None);
         }
 
-        debug!(target: "nitox", "codec buffer is {:?}", buf);
+        debug!(target: "nitox", "next index: {}", self.next_index);
+
+        //debug!(target: "nitox", "codec buffer is {:?}", buf);
         // Let's check if we find a blank space at the beginning
         if let Some(command_offset) = buf[self.next_index..]
             .iter()
             .position(|b| *b == b' ' || *b == b'\t' || *b == b'\r')
         {
             let command_end = self.next_index + command_offset;
-            debug!(target: "nitox", "codec detected command name {:?}", &buf[..command_end]);
+            debug!(target: "nitox", "command end: {}", command_end);
+            //debug!(target: "nitox", "codec detected command name {:?}", &buf[..command_end]);
+            if !Op::command_exists(&buf[..command_end]) {
+                debug!(target: "nitox", "command was incomplete");
+                return Ok(None);
+            }
 
             if let Some(command_body_offset) = buf[command_end..].windows(2).position(|w| w == b"\r\n") {
                 let mut end_buf_pos = command_end + command_body_offset + 2;
@@ -60,29 +67,19 @@ impl Decoder for OpCodec {
                         end_buf_pos += new_end + 2;
                     } else {
                         debug!(target: "nitox", "command was incomplete");
-                        self.next_index = buf.len();
                         return Ok(None);
                     }
                 }
 
                 debug!(target: "nitox", "codec detected command body {:?}", &buf[..end_buf_pos]);
-                if !Op::command_exists(&buf[..command_end]) {
-                    debug!(target: "nitox", "command was incomplete");
-                    self.next_index = buf.len();
-                    return Ok(None);
-                }
 
+                let cmd_buf = buf.split_to(end_buf_pos);
                 debug!(target: "nitox", "buffer now contains {:?}", buf);
                 self.next_index = 0;
-                match Op::from_bytes(buf[..command_end].into(), command_end) {
-                    Err(CommandError::IncompleteCommandError) => {
-                        debug!(target: "nitox", "command was incomplete");
-                        self.next_index = buf.len();
-                        Ok(None)
-                    }
+
+                match Op::from_bytes(cmd_buf.freeze(), command_end) {
                     Ok(op) => {
                         debug!(target: "nitox", "codec parsed command {:#?}", op);
-                        let _ = buf.split_to(end_buf_pos);
                         Ok(Some(op))
                     }
                     Err(e) => {
@@ -92,8 +89,7 @@ impl Decoder for OpCodec {
                 }
             } else {
                 debug!(target: "nitox", "command was incomplete");
-                self.next_index = buf.len();
-                return Ok(None);
+                Ok(None)
             }
         } else {
             // First blank not found yet, continuing

src/protocol/client/pub_cmd.rs

@@ -35,8 +35,9 @@ impl Command for PubCommand {
 
     fn into_vec(self) -> Result<Bytes, CommandError> {
         let (rt_len, rt) = self.reply_to.map_or((0, "".into()), |rp| (rp.len() + 1, rp));
+        // Computes the string length of the payload_len by dividing the number par ln(10)
         let size_len = ((self.payload.len() + 1) as f64 / std::f64::consts::LN_10).ceil() as usize;
-        let len = 9 + self.subject.len() + rt_len + size_len;
+        let len = 9 + self.subject.len() + rt_len + size_len + self.payload.len();
 
         let mut bytes = BytesMut::with_capacity(len);
         bytes.put("PUB\t");

src/protocol/client/unsub_cmd.rs

@@ -32,6 +32,7 @@ impl Command for UnsubCommand {
     const CMD_NAME: &'static [u8] = b"UNSUB";
 
     fn into_vec(self) -> Result<Bytes, CommandError> {
+        // Computes the string length of the payload_len by dividing the number par ln(10)
         let (mm_len, mm) = self.max_msgs.map_or((0, 0), |mm| {
             (((mm + 1) as f64 / std::f64::consts::LN_10).ceil() as usize, mm)
         });