Merge pull request #11 from pleshevskiy/bug-8

fix!: handle sonic server-side errors

src/channels/mod.rs

@@ -101,12 +101,13 @@ impl SonicStream {
         let mut reader = BufReader::with_capacity(self.max_buffer_size, &self.stream);
         let mut message = String::new();
 
-        let mut lines_read = 0;
+        for _ in 0..max_read_lines {
-        while lines_read < max_read_lines {
             reader
                 .read_line(&mut message)
                 .map_err(|_| Error::new(ErrorKind::ReadStream))?;
-            lines_read += 1;
+            if message.starts_with("ERR ") {
+                break;
+            }
         }
 
         Ok(message)
@@ -115,8 +116,14 @@ impl SonicStream {
     pub(crate) fn run_command<SC: StreamCommand>(&self, command: SC) -> Result<SC::Response> {
         self.write(&command)?;
         let message = self.read(SC::READ_LINES_COUNT)?;
+        if let Some(error) = message.strip_prefix("ERR ") {
+            Err(Error::new(ErrorKind::SonicServer(Box::leak(
+                error.to_owned().into_boxed_str(),
+            ))))
+        } else {
             command.receive(message)
         }
+    }
 
     fn connect<A: ToSocketAddrs>(addr: A) -> Result<Self> {
         let stream =
@@ -212,3 +219,15 @@ pub trait SonicChannel {
         A: ToSocketAddrs,
         S: ToString;
 }
+
+#[cfg(test)]
+mod tests {
+    use super::ChannelMode;
+
+    #[test]
+    fn format_channel_enums() {
+        assert_eq!(format!("{}", ChannelMode::Search), String::from("search"));
+        assert_eq!(format!("{}", ChannelMode::Ingest), String::from("ingest"));
+        assert_eq!(format!("{}", ChannelMode::Control), String::from("control"));
+    }
+}

src/commands/count.rs

@@ -28,12 +28,12 @@ impl StreamCommand for CountCommand<'_> {
         if message.starts_with("RESULT ") {
             let count = message.split_whitespace().last().unwrap_or_default();
             count.parse().map_err(|_| {
-                Error::new(ErrorKind::QueryResponseError(
+                Error::new(ErrorKind::QueryResponse(
                     "Cannot parse count of count method response to usize",
                 ))
             })
         } else {
-            Err(Error::new(ErrorKind::WrongSonicResponse))
+            Err(Error::new(ErrorKind::WrongResponse))
         }
     }
 }

src/commands/flush.rs

@@ -28,12 +28,12 @@ impl StreamCommand for FlushCommand<'_> {
         if message.starts_with("RESULT ") {
             let count = message.split_whitespace().last().unwrap_or_default();
             count.parse().map_err(|_| {
-                Error::new(ErrorKind::QueryResponseError(
+                Error::new(ErrorKind::QueryResponse(
                     "Cannot parse count of flush method response to usize",
                 ))
             })
         } else {
-            Err(Error::new(ErrorKind::WrongSonicResponse))
+            Err(Error::new(ErrorKind::WrongResponse))
         }
     }
 }

src/commands/ping.rs

@@ -15,7 +15,7 @@ impl StreamCommand for PingCommand {
         if message == "PONG\r\n" {
             Ok(true)
         } else {
-            Err(Error::new(ErrorKind::WrongSonicResponse))
+            Err(Error::new(ErrorKind::WrongResponse))
         }
     }
 }

src/commands/pop.rs

@@ -25,12 +25,12 @@ impl StreamCommand for PopCommand<'_> {
         if message.starts_with("RESULT ") {
             let count = message.split_whitespace().last().unwrap_or_default();
             count.parse().map_err(|_| {
-                Error::new(ErrorKind::QueryResponseError(
+                Error::new(ErrorKind::QueryResponse(
                     "Cannot parse count of pop method response to usize",
                 ))
             })
         } else {
-            Err(Error::new(ErrorKind::WrongSonicResponse))
+            Err(Error::new(ErrorKind::WrongResponse))
         }
     }
 }

src/commands/push.rs

@@ -31,12 +31,8 @@ impl StreamCommand for PushCommand<'_> {
     fn receive(&self, message: String) -> Result<Self::Response> {
         if message == "OK\r\n" {
             Ok(true)
-        } else if message.starts_with("ERR ") {
-            Err(Error::new(ErrorKind::QueryResponseError(Box::leak(
-                message.into_boxed_str(),
-            ))))
         } else {
-            Err(Error::new(ErrorKind::WrongSonicResponse))
+            Err(Error::new(ErrorKind::WrongResponse))
         }
     }
 }

src/commands/query.rs

@@ -43,7 +43,7 @@ impl StreamCommand for QueryCommand<'_> {
 
         if let Some(caps) = RE.captures(&message) {
             if caps["pending_query_id"] != caps["event_query_id"] {
-                Err(Error::new(ErrorKind::QueryResponseError(
+                Err(Error::new(ErrorKind::QueryResponse(
                     "Pending id and event id don't match",
                 )))
             } else if caps["objects"].is_empty() {
@@ -55,7 +55,7 @@ impl StreamCommand for QueryCommand<'_> {
                     .collect())
             }
         } else {
-            Err(Error::new(ErrorKind::WrongSonicResponse))
+            Err(Error::new(ErrorKind::WrongResponse))
         }
     }
 }

src/commands/quit.rs

@@ -15,7 +15,7 @@ impl StreamCommand for QuitCommand {
         if message.starts_with("ENDED ") {
             Ok(true)
         } else {
-            Err(Error::new(ErrorKind::WrongSonicResponse))
+            Err(Error::new(ErrorKind::WrongResponse))
         }
     }
 }

src/commands/suggest.rs

@@ -38,10 +38,10 @@ impl StreamCommand for SuggestCommand<'_> {
         }
 
         match RE.captures(&message) {
-            None => Err(Error::new(ErrorKind::WrongSonicResponse)),
+            None => Err(Error::new(ErrorKind::WrongResponse)),
             Some(caps) => {
                 if caps["pending_suggest_id"] != caps["event_suggest_id"] {
-                    Err(Error::new(ErrorKind::QueryResponseError(
+                    Err(Error::new(ErrorKind::QueryResponse(
                         "Pending id and event id don't match",
                     )))
                 } else if caps["words"].is_empty() {

src/commands/trigger.rs

@@ -41,7 +41,7 @@ impl StreamCommand for TriggerCommand<'_> {
         if message == "OK\r\n" {
             Ok(true)
         } else {
-            Err(Error::new(ErrorKind::WrongSonicResponse))
+            Err(Error::new(ErrorKind::WrongResponse))
         }
     }
 }

src/lib.rs

@@ -99,17 +99,3 @@ pub use channels::*;
 #[macro_use]
 extern crate lazy_static;
 extern crate regex;
-
-#[cfg(test)]
-mod tests {
-    use crate::channels::ChannelMode;
-
-    #[test]
-    fn format_channel_enums() {
-        assert_eq!(format!("{}", ChannelMode::Search), String::from("search"));
-        assert_eq!(format!("{}", ChannelMode::Ingest), String::from("ingest"));
-        assert_eq!(format!("{}", ChannelMode::Control), String::from("control"));
-    }
-
-    //TODO: write tests with sonic server
-}

src/result.rs

@@ -47,14 +47,17 @@ pub enum ErrorKind {
     RunCommand,
 
     /// Error in query response with additional message.
-    QueryResponseError(&'static str),
+    QueryResponse(&'static str),
 
     /// Response from sonic server are wrong! Actually it may happen if you use
     /// unsupported sonic backend version. Please write issue to the github repo.
-    WrongSonicResponse,
+    WrongResponse,
 
     /// You cannot run the command in current channel.
     UnsupportedCommand((&'static str, Option<ChannelMode>)),
+
+    /// This error appears if the error occurred on the server side
+    SonicServer(&'static str),
 }
 
 impl fmt::Display for Error {
@@ -65,11 +68,11 @@ impl fmt::Display for Error {
             ErrorKind::ReadStream => write!(f, "Cannot read sonic response from stream"),
             ErrorKind::SwitchMode => write!(f, "Cannot switch channel mode"),
             ErrorKind::RunCommand => write!(f, "Cannot run command in current mode"),
-            ErrorKind::QueryResponseError(message) => {
+            ErrorKind::QueryResponse(message) => {
                 write!(f, "Error in query response: {}", message)
             }
-            ErrorKind::WrongSonicResponse => {
+            ErrorKind::WrongResponse => {
-                write!(f, "Sonic response are wrong. Please write issue to github.")
+                write!(f, "Client cannot parse response from sonic server. Please write an issue to github (https://github.com/pleshevskiy/sonic-channel).")
             }
             ErrorKind::UnsupportedCommand((command_name, channel_mode)) => {
                 if let Some(channel_mode) = channel_mode {
@@ -86,6 +89,7 @@ impl fmt::Display for Error {
                     )
                 }
             }
+            ErrorKind::SonicServer(message) => write!(f, "Sonic Server-side error: {}", message),
         }
     }
 }