add server pub/sub tests, fix pub/sub response (#27)

mini-redis server responses on PUB/SUB commands did not match real
redis.

Author: carllerche

src/client.rs

@@ -143,7 +143,7 @@ impl Client {
             let response = self.read_response().await?;
             match response {
                 Frame::Array(ref frame) => match frame.as_slice() {
-                    [subscribe, schannel]
+                    [subscribe, schannel, ..]
                         if subscribe.to_string() == "subscribe"
                             && &schannel.to_string() == channel =>
                     {
@@ -235,7 +235,7 @@ impl Subscriber {
             let response = self.read_response().await?;
             match response {
                 Frame::Array(ref frame) => match frame.as_slice() {
-                    [subscribe, schannel]
+                    [subscribe, schannel, ..]
                         if &subscribe.to_string() == "subscribe"
                             && &schannel.to_string() == channel =>
                     {
@@ -277,7 +277,7 @@ impl Subscriber {
             let response = self.read_response().await?;
             match response {
                 Frame::Array(ref frame) => match frame.as_slice() {
-                    [unsubscribe, uchannel] if &unsubscribe.to_string() == "unsubscribe" => {
+                    [unsubscribe, uchannel, ..] if &unsubscribe.to_string() == "unsubscribe" => {
                         //unsubscribed channel should exist in the subscribed list at this point
                         if self.subscribed_channels.remove(&uchannel.to_string()) == false {
                             return Err(response.to_error());

src/cmd/mod.rs

@@ -26,7 +26,7 @@ pub(crate) enum Command {
 }
 
 impl Command {
-    pub(crate) fn from_frame(frame: Frame) -> Result<Command, ParseError> {
+    pub(crate) fn from_frame(frame: Frame) -> crate::Result<Command> {
         let mut parse = Parse::new(frame)?;
 
         let command_name = parse.next_string()?.to_lowercase();

src/cmd/subscribe.rs

@@ -68,6 +68,7 @@ impl Subscribe {
                 let mut response = Frame::array();
                 response.push_bulk(Bytes::from_static(b"subscribe"));
                 response.push_bulk(Bytes::copy_from_slice(channel.as_bytes()));
+                response.push_int(subscriptions.len().saturating_add(1) as u64);
 
                 // Subscribe to channel
                 let rx = db.subscribe(channel.clone());
@@ -130,6 +131,7 @@ impl Subscribe {
                                 let mut response = Frame::array();
                                 response.push_bulk(Bytes::from_static(b"unsubscribe"));
                                 response.push_bulk(Bytes::copy_from_slice(channel.as_bytes()));
+                                response.push_int(subscriptions.len() as u64);
 
                                 dst.write_frame(&response).await?;
                             }

src/frame.rs

@@ -21,7 +21,7 @@ pub(crate) enum Error {
     Incomplete,
 
     /// Invalid message encoding
-    Invalid,
+    Other(crate::Error),
 }
 
 impl Frame {
@@ -30,7 +30,7 @@ impl Frame {
         Frame::Array(vec![])
     }
 
-    /// Push a "bulk" frame into the array. `self` must be an Array frame
+    /// Push a "bulk" frame into the array. `self` must be an Array frame.
     ///
     /// # Panics
     ///
@@ -44,6 +44,20 @@ impl Frame {
         }
     }
 
+    /// Push an "integer" frame into the array. `self` must be an Array frame.
+    ///
+    /// # Panics
+    ///
+    /// panics if `self` is not an array
+    pub(crate) fn push_int(&mut self, value: u64) {
+        match self {
+            Frame::Array(vec) => {
+                vec.push(Box::new(Frame::Integer(value)));
+            }
+            _ => panic!("not an array frame"),
+        }
+    }
+
     /// Checks if an entire message can be decoded from `src`
     pub(crate) fn check(src: &mut Cursor<&[u8]>) -> Result<(), Error> {
         match get_u8(src)? {
@@ -80,7 +94,7 @@ impl Frame {
 
                 Ok(())
             }
-            _ => Err(Error::Invalid),
+            actual => Err(format!("protocol error; invalid frame type byte `{}`", actual).into()),
         }
     }
 
@@ -114,7 +128,7 @@ impl Frame {
                     let line = get_line(src)?;
 
                     if line != b"-1" {
-                        return Err(Error::Invalid);
+                        return Err("protocol error; invalid frame format".into());
                     }
 
                     Ok(Frame::Null)
@@ -213,7 +227,7 @@ fn get_decimal(src: &mut Cursor<&[u8]>) -> Result<u64, Error> {
 
     let line = get_line(src)?;
 
-    atoi::<u64>(line).ok_or(Error::Invalid)
+    atoi::<u64>(line).ok_or_else(|| "protocol error; invalid frame format".into())
 }
 
 /// Find a line
@@ -236,15 +250,27 @@ fn get_line<'a>(src: &mut Cursor<&'a [u8]>) -> Result<&'a [u8], Error> {
     Err(Error::Incomplete)
 }
 
+impl From<String> for Error {
+    fn from(src: String) -> Error {
+        Error::Other(src.into())
+    }
+}
+
+impl From<&str> for Error {
+    fn from(src: &str) -> Error {
+        src.to_string().into()
+    }
+}
+
 impl From<FromUtf8Error> for Error {
     fn from(_src: FromUtf8Error) -> Error {
-        unimplemented!();
+        "protocol error; invalid frame format".into()
     }
 }
 
 impl From<TryFromIntError> for Error {
     fn from(_src: TryFromIntError) -> Error {
-        unimplemented!();
+        "protocol error; invalid frame format".into()
     }
 }
 
@@ -254,7 +280,7 @@ impl fmt::Display for Error {
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
         match self {
             Error::Incomplete => "stream ended early".fmt(fmt),
-            Error::Invalid => "invalid frame format".fmt(fmt),
+            Error::Other(err) => err.fmt(fmt),
         }
     }
 }

src/parse.rs

@@ -1,7 +1,7 @@
 use crate::Frame;
 
 use bytes::Bytes;
-use std::{error, fmt, io, str, vec};
+use std::{fmt, str, vec};
 
 /// Utility for parsing a command
 #[derive(Debug)]
@@ -12,14 +12,14 @@ pub(crate) struct Parse {
 #[derive(Debug)]
 pub(crate) enum ParseError {
     EndOfStream,
-    Invalid,
+    Other(crate::Error),
 }
 
 impl Parse {
     pub(crate) fn new(frame: Frame) -> Result<Parse, ParseError> {
         let array = match frame {
             Frame::Array(array) => array,
-            _ => return Err(ParseError::Invalid),
+            frame => return Err(format!("protocol error; expected array, got {:?}", frame).into()),
         };
 
         Ok(Parse {
@@ -39,27 +39,29 @@ impl Parse {
             Frame::Simple(s) => Ok(s),
             Frame::Bulk(data) => str::from_utf8(&data[..])
                 .map(|s| s.to_string())
-                .map_err(|_| ParseError::Invalid),
-            _ => Err(ParseError::Invalid),
+                .map_err(|_| "protocol error; invalid string".into()),
+            frame => Err(format!("protocol error; expected simple frame or bulk frame, got {:?}", frame).into()),
         }
     }
 
     pub(crate) fn next_bytes(&mut self) -> Result<Bytes, ParseError> {
         match self.next()? {
             Frame::Simple(s) => Ok(Bytes::from(s.into_bytes())),
             Frame::Bulk(data) => Ok(data),
-            _ => Err(ParseError::Invalid),
+            frame => Err(format!("protocol error; expected simple frame or bulk frame, got {:?}", frame).into()),
         }
     }
 
     pub(crate) fn next_int(&mut self) -> Result<u64, ParseError> {
         use atoi::atoi;
 
+        const MSG: &str = "protocol error; invalid number";
+
         match self.next()? {
             Frame::Integer(v) => Ok(v),
-            Frame::Simple(data) => atoi::<u64>(data.as_bytes()).ok_or(ParseError::Invalid),
-            Frame::Bulk(data) => atoi::<u64>(&data).ok_or(ParseError::Invalid),
-            _ => Err(ParseError::Invalid),
+            Frame::Simple(data) => atoi::<u64>(data.as_bytes()).ok_or_else(|| MSG.into()),
+            Frame::Bulk(data) => atoi::<u64>(&data).ok_or_else(|| MSG.into()),
+            frame => Err(format!("protocol error; expected int frame but got {:?}", frame).into()),
         }
     }
 
@@ -68,29 +70,33 @@ impl Parse {
         if self.parts.next().is_none() {
             Ok(())
         } else {
-            Err(ParseError::Invalid)
+            Err("protocol error; expected end of frame, but there was more".into())
         }
     }
 }
 
-impl From<ParseError> for io::Error {
-    fn from(src: ParseError) -> io::Error {
-        io::Error::new(io::ErrorKind::Other, format!("{}", src))
+impl From<String> for ParseError {
+    fn from(src: String) -> ParseError {
+        ParseError::Other(src.into())
+    }
+}
+
+impl From<&str> for ParseError {
+    fn from(src: &str) -> ParseError {
+        src.to_string().into()
     }
 }
 
 impl fmt::Display for ParseError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        let msg = match self {
-            ParseError::EndOfStream => "end of stream".to_string(),
-            ParseError::Invalid => "invalid".to_string(),
-        };
-        write!(f, "{}", &msg)
+        match self {
+            ParseError::EndOfStream => {
+                "protocol error; unexpected end of stream".fmt(f)
+            }
+            ParseError::Other(err) => err.fmt(f),
+        }
     }
 }
 
 impl std::error::Error for ParseError {
-    fn source(&self) -> Option<&(dyn error::Error + 'static)> {
-        None
-    }
 }