Update the tokio codec to use new features.

Author: jmaygarden

examples/tokio-serial.rs

@@ -1,5 +1,6 @@
 use futures::{ready, SinkExt, StreamExt};
 use serialport::SerialPort;
+use slip_codec::tokio::SlipCodec;
 use std::io::{Read, Write};
 use std::task::Context;
 use tokio::io::unix::AsyncFd;
@@ -80,7 +81,7 @@ impl AsyncWrite for AsyncTTYPort {
 }
 
 async fn run_source(port: AsyncTTYPort) {
-    let mut sink = tokio_util::codec::Framed::new(port, slip_codec::SlipCodec::new());
+    let mut sink = tokio_util::codec::Framed::new(port, SlipCodec::new());
 
     for message in ["foo", "bar", "baz"].iter() {
         let message = message.to_string().into();
@@ -91,15 +92,14 @@ async fn run_source(port: AsyncTTYPort) {
 }
 
 async fn run_sink(port: AsyncTTYPort) {
-    let mut source = tokio_util::codec::Framed::new(port, slip_codec::SlipCodec::new());
+    let mut source = tokio_util::codec::Framed::new(port, SlipCodec::new());
 
     loop {
-        match source.next().await {
-            Some(result) => match result {
+        if let Some(result)  = source.next().await {
+            match result {
                 Ok(message) => println!("recv {:?}", message),
                 Err(_) => break,
-            },
-            None => (),
+            }
         }
     }
 }

src/tokio/codec.rs

@@ -0,0 +1,73 @@
+use super::{SlipDecoder, SlipEncoder};
+use crate::{SlipError, MAX_PACKET_SIZE};
+use bytes::{Bytes, BytesMut};
+use tokio_util::codec::{Decoder, Encoder};
+
+pub struct SlipCodec {
+    decoder: SlipDecoder,
+    encoder: SlipEncoder,
+}
+
+pub struct SlipCodecBuilder {
+    begin_with_end: bool,
+    capacity: usize,
+}
+
+impl SlipCodec {
+    pub fn new() -> Self {
+        Self {
+            decoder: SlipDecoder::default(),
+            encoder: SlipEncoder::default(),
+        }
+    }
+
+    pub fn builder() -> SlipCodecBuilder {
+        SlipCodecBuilder {
+            begin_with_end: true,
+            capacity: MAX_PACKET_SIZE,
+        }
+    }
+}
+
+impl Decoder for SlipCodec {
+    type Item = BytesMut;
+    type Error = SlipError;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<BytesMut>, Self::Error> {
+        self.decoder.decode(src)
+    }
+}
+
+impl Encoder<Bytes> for SlipCodec {
+    type Error = SlipError;
+
+    fn encode(&mut self, item: Bytes, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        self.encoder.encode(item, dst).map_err(SlipError::ReadError)
+    }
+}
+
+impl Default for SlipCodec {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl SlipCodecBuilder {
+    pub fn begin_with_end(self, begin_with_end: bool) -> Self {
+        Self {
+            begin_with_end,
+            ..self
+        }
+    }
+
+    pub fn capacity(self, capacity: usize) -> Self {
+        Self { capacity, ..self }
+    }
+
+    pub fn build(self) -> SlipCodec {
+        SlipCodec {
+            decoder: SlipDecoder::with_capacity(self.capacity),
+            encoder: SlipEncoder::new(self.begin_with_end),
+        }
+    }
+}

src/tokio/decoder.rs

@@ -1,102 +1,47 @@
-use crate::{SlipCodecError, END, ESC, ESC_END, ESC_ESC};
-
+use crate::{SlipError, MAX_PACKET_SIZE};
 use bytes::{Buf, BufMut, BytesMut};
 use tokio_util::codec::Decoder;
 
-enum State {
-    Normal,
-    Error,
-    Escape,
-}
-
 /// SLIP decoding context
 pub struct SlipDecoder {
     buf: BytesMut,
-    capacity: Option<usize>,
-    state: State,
+    capacity: usize,
+    inner: crate::SlipDecoder,
 }
 
-const INITIAL_CAPACITY: usize = 1024;
-
 impl SlipDecoder {
-    /// Creates a new context without a maximum buffer size.
-    pub fn new() -> Self {
-        Self {
-            buf: BytesMut::with_capacity(INITIAL_CAPACITY),
-            capacity: None,
-            state: State::Normal,
-        }
-    }
-
     /// Creates a new context with the given maximum buffer size.
     pub fn with_capacity(capacity: usize) -> Self {
         Self {
             buf: BytesMut::with_capacity(capacity),
-            capacity: Some(capacity),
-            state: State::Normal,
+            capacity,
+            inner: Default::default(),
         }
     }
-
-    fn push(self: &mut Self, value: u8) -> Result<(), SlipCodecError> {
-        if let Some(capacity) = self.capacity {
-            if self.buf.len() == capacity {
-                return Err(SlipCodecError::OversizedPacket);
-            }
-        }
-
-        self.buf.put_u8(value);
-
-        Ok(())
-    }
 }
 
 impl Decoder for SlipDecoder {
     type Item = BytesMut;
-    type Error = SlipCodecError;
-
-    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<BytesMut>, SlipCodecError> {
-        while src.has_remaining() {
-            let value = src.get_u8();
-
-            match self.state {
-                State::Normal => match value {
-                    END => {
-                        if self.buf.has_remaining() {
-                            return Ok(Some(self.buf.split()));
-                        }
-                    }
-                    ESC => {
-                        self.state = State::Escape;
-                    }
-                    _ => {
-                        self.push(value)?;
-                    }
-                },
-                State::Error => {
-                    if value == END {
-                        self.buf.clear();
-                        self.state = State::Normal;
-                    }
-                }
-                State::Escape => match value {
-                    ESC_END => {
-                        self.push(END)?;
-                        self.state = State::Normal;
-                    }
-                    ESC_ESC => {
-                        self.push(ESC)?;
-                        self.state = State::Normal;
-                    }
-                    _ => {
-                        self.state = State::Error;
-
-                        return Err(SlipCodecError::FramingError);
-                    }
-                },
-            }
+    type Error = SlipError;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        let src = &mut src.reader();
+        let dst = {
+            self.buf.reserve(self.capacity);
+            &mut (&mut self.buf).limit(self.capacity).writer()
+        };
+
+        match self.inner.decode(src, dst) {
+            Ok(len) => Ok(Some(self.buf.split_to(len))),
+            Err(SlipError::EndOfStream) => Ok(None),
+            Err(e) => Err(e),
         }
+    }
+}
 
-        Ok(None)
+impl Default for SlipDecoder {
+    fn default() -> Self {
+        Self::with_capacity(MAX_PACKET_SIZE)
     }
 }
 
@@ -108,7 +53,7 @@ mod tests {
     fn empty_decode() {
         const INPUT: [u8; 2] = [0xc0, 0xc0];
 
-        let mut slip = SlipDecoder::new();
+        let mut slip = SlipDecoder::default();
         let mut buf = BytesMut::from(&INPUT[..]);
         let res = slip.decode(&mut buf).unwrap();
         assert!(res.is_none());
@@ -120,9 +65,10 @@ mod tests {
         const INPUT: [u8; 7] = [0xc0, 0x01, 0x02, 0x03, 0x04, 0x05, 0xc0];
         const DATA: [u8; 5] = [0x01, 0x02, 0x03, 0x04, 0x05];
 
-        let mut slip = SlipDecoder::new();
+        let mut slip = SlipDecoder::default();
         let mut buf = BytesMut::from(&INPUT[..]);
         let buf = slip.decode(&mut buf).unwrap().unwrap();
+        eprintln!("{}:{}: {:?}", file!(), line!(), buf);
         assert_eq!(DATA.len(), buf.len());
         assert_eq!(&DATA[..], &buf);
     }
@@ -133,7 +79,7 @@ mod tests {
         const INPUT: [u8; 6] = [0xc0, 0x01, 0xdb, 0xdc, 0x03, 0xc0];
         const DATA: [u8; 3] = [0x01, 0xc0, 0x03];
 
-        let mut slip = SlipDecoder::new();
+        let mut slip = SlipDecoder::default();
         let mut buf = BytesMut::from(&INPUT[..]);
         let buf = slip.decode(&mut buf).unwrap().unwrap();
         assert_eq!(DATA.len(), buf.len());
@@ -146,7 +92,7 @@ mod tests {
         const INPUT: [u8; 6] = [0xc0, 0x01, 0xdb, 0xdd, 0x03, 0xc0];
         const DATA: [u8; 3] = [0x01, 0xdb, 0x03];
 
-        let mut slip = SlipDecoder::new();
+        let mut slip = SlipDecoder::default();
         let mut buf = BytesMut::from(&INPUT[..]);
         let buf = slip.decode(&mut buf).unwrap().unwrap();
         assert_eq!(DATA.len(), buf.len());
@@ -159,7 +105,7 @@ mod tests {
         const INPUT_2: [u8; 6] = [0x05, 0x06, 0x07, 0x08, 0x09, 0xc0];
         const DATA: [u8; 10] = [0x01, 0x02, 0x03, 0x04, 0x05, 0x05, 0x06, 0x07, 0x08, 0x09];
 
-        let mut slip = SlipDecoder::new();
+        let mut slip = SlipDecoder::default();
         let mut buf = BytesMut::from(&INPUT_1[..]);
 
         {

src/tokio/encoder.rs

@@ -1,62 +1,56 @@
-use crate::{SlipCodecError, END, ESC, ESC_END, ESC_ESC};
-
-use bytes::{Buf, BufMut, Bytes, BytesMut};
+use bytes::{BufMut, Bytes, BytesMut};
 use tokio_util::codec::Encoder;
 
 /// SLIP encoder context
-pub struct SlipEncoder {}
+pub struct SlipEncoder {
+    inner: crate::SlipEncoder,
+}
 
 impl SlipEncoder {
     /// Creates a new encoder context
-    pub fn new() -> Self {
-        Self {}
+    pub fn new(begin_with_end: bool) -> Self {
+        Self {
+            inner: crate::SlipEncoder::new(begin_with_end),
+        }
     }
 }
 
 impl Encoder<Bytes> for SlipEncoder {
-    type Error = SlipCodecError;
-
-    fn encode(&mut self, mut item: Bytes, dst: &mut BytesMut) -> Result<(), SlipCodecError> {
-        dst.reserve(item.len());
-
-        dst.put_u8(END);
-
-        while item.has_remaining() {
-            let value = item.get_u8();
-
-            match value {
-                END => {
-                    dst.put_u8(ESC);
-                    dst.put_u8(ESC_END);
-                }
-                ESC => {
-                    dst.put_u8(ESC);
-                    dst.put_u8(ESC_ESC);
-                }
-                _ => {
-                    dst.put_u8(value);
-                }
-            }
-        }
+    type Error = std::io::Error;
 
-        dst.put_u8(END);
+    fn encode(&mut self, item: Bytes, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        self.inner
+            .encode(item.as_ref(), &mut dst.writer())
+            .map(|_| ())
+    }
+}
 
-        Ok(())
+impl Default for SlipEncoder {
+    fn default() -> Self {
+        Self::new(true)
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::{END, ESC, ESC_END, ESC_ESC};
 
     #[test]
     fn empty_encode() {
         const EXPECTED: [u8; 2] = [0xc0, 0xc0];
-        let mut output = BytesMut::new();
 
-        let mut slip = SlipEncoder::new();
+        // default is to begin and end with END tokens
+        let mut output = BytesMut::new();
+        let mut slip = SlipEncoder::default();
         slip.encode(Bytes::new(), &mut output).unwrap();
         assert_eq!(&EXPECTED[..], &output);
+
+        // override to only use END token to terminate packet
+        let mut output = BytesMut::new();
+        let mut slip = SlipEncoder::new(false);
+        slip.encode(Bytes::new(), &mut output).unwrap();
+        assert_eq!(&EXPECTED[..1], &output);
     }
 
     #[test]
@@ -65,7 +59,7 @@ mod tests {
         const EXPECTED: [u8; 6] = [0xc0, 0x01, ESC, ESC_ESC, 0x03, 0xc0];
         let mut output = BytesMut::new();
 
-        let mut slip = SlipEncoder::new();
+        let mut slip = SlipEncoder::default();
         slip.encode(Bytes::from(&INPUT[..]), &mut output).unwrap();
         assert_eq!(&EXPECTED[..], &output);
     }
@@ -76,7 +70,7 @@ mod tests {
         const EXPECTED: [u8; 6] = [0xc0, 0x01, ESC, ESC_END, 0x03, 0xc0];
         let mut output = BytesMut::new();
 
-        let mut slip = SlipEncoder::new();
+        let mut slip = SlipEncoder::default();
         slip.encode(Bytes::from(&INPUT[..]), &mut output).unwrap();
         assert_eq!(&EXPECTED[..], &output);
     }