diff --git a/src/channels/mod.rs b/src/channels/mod.rs index 91fe5b9..0158d36 100644 --- a/src/channels/mod.rs +++ b/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; - while lines_read < max_read_lines { + for _ in 0..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,7 +116,13 @@ impl SonicStream { pub(crate) fn run_command(&self, command: SC) -> Result { self.write(&command)?; let message = self.read(SC::READ_LINES_COUNT)?; - command.receive(message) + 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(addr: A) -> Result { @@ -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")); + } +} diff --git a/src/commands/count.rs b/src/commands/count.rs index 93744a9..35a031b 100644 --- a/src/commands/count.rs +++ b/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)) } } } diff --git a/src/commands/flush.rs b/src/commands/flush.rs index c03efa7..5d29015 100644 --- a/src/commands/flush.rs +++ b/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)) } } } diff --git a/src/commands/ping.rs b/src/commands/ping.rs index 5c09bbc..3eb3030 100644 --- a/src/commands/ping.rs +++ b/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)) } } } diff --git a/src/commands/pop.rs b/src/commands/pop.rs index 878c31f..dd52549 100644 --- a/src/commands/pop.rs +++ b/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)) } } } diff --git a/src/commands/push.rs b/src/commands/push.rs index f3d9ab2..2c2ee38 100644 --- a/src/commands/push.rs +++ b/src/commands/push.rs @@ -31,12 +31,8 @@ impl StreamCommand for PushCommand<'_> { fn receive(&self, message: String) -> Result { 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)) } } } diff --git a/src/commands/query.rs b/src/commands/query.rs index f3ebc9d..8823a7c 100644 --- a/src/commands/query.rs +++ b/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)) } } } diff --git a/src/commands/quit.rs b/src/commands/quit.rs index adfc1d7..30fc6f1 100644 --- a/src/commands/quit.rs +++ b/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)) } } } diff --git a/src/commands/suggest.rs b/src/commands/suggest.rs index 27cb645..66a4f88 100644 --- a/src/commands/suggest.rs +++ b/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() { diff --git a/src/commands/trigger.rs b/src/commands/trigger.rs index 57a8f4e..5067fe6 100644 --- a/src/commands/trigger.rs +++ b/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)) } } } diff --git a/src/lib.rs b/src/lib.rs index ee253a8..4f38a74 100644 --- a/src/lib.rs +++ b/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 -} diff --git a/src/result.rs b/src/result.rs index 12b27c0..b86da14 100644 --- a/src/result.rs +++ b/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)), + + /// 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 => { - write!(f, "Sonic response are wrong. Please write issue to github.") + ErrorKind::WrongResponse => { + 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), } } }