Add sentence I/O to blocking ConnectionStream

rups/Cargo.toml

@@ -21,6 +21,10 @@ webpki-roots = { version = "0.21", optional = true }
 tokio = { version = "1", optional = true, features = ["net", "io-util", "rt"] }
 tokio-rustls = { version = "0.22", optional = true }
 
+[dev-dependencies]
+mockstream = "0.0.3"
+tokio-mockstream = "1.1.0"
+
 [features]
 default = []
 ssl = ["rustls", "rustls/dangerous_configuration", "webpki", "webpki-roots"]

rups/src/blocking/client.rs

@@ -0,0 +1,12 @@
+use crate::blocking::stream::ConnectionStream;
+use crate::Config;
+
+/// A synchronous NUT client.
+pub struct Client {
+    /// The client configuration.
+    config: Config,
+    /// The client connection.
+    stream: ConnectionStream,
+}
+
+impl Client {}

rups/src/blocking/mod.rs

@@ -5,6 +5,7 @@ use crate::blocking::stream::ConnectionStream;
 use crate::cmd::{Command, Response};
 use crate::{Config, Error, Host, NutError};
 
+mod client;
 mod stream;
 
 /// A blocking NUT client connection.

rups/src/blocking/stream.rs

@@ -1,4 +1,7 @@
-use std::io::{Read, Write};
+use crate::proto::util::{join_sentence, split_sentence};
+use crate::proto::Sentence;
+use crate::{Error, NutError};
+use std::io::{BufRead, BufReader, Read, Write};
 use std::net::TcpStream;
 
 /// A wrapper for various synchronous stream types.
@@ -13,6 +16,10 @@ pub enum ConnectionStream {
     /// A server stream wrapped with SSL using `rustls`.
     #[cfg(feature = "ssl")]
     SslServer(Box<rustls::StreamOwned<rustls::ServerSession, ConnectionStream>>),
+
+    /// A mock stream, used for testing.
+    #[cfg(test)]
+    Mock(mockstream::SharedMockStream),
 }
 
 impl ConnectionStream {
@@ -37,6 +44,82 @@ impl ConnectionStream {
             rustls::StreamOwned::new(session, self),
         )))
     }
+
+    /// Writes a sentence on the current stream.
+    pub fn write_sentence<T: Sentence>(&mut self, sentence: &T) -> crate::Result<()> {
+        let encoded = sentence.encode();
+        let joined = join_sentence(encoded);
+        self.write_literal(&joined)?;
+        self.flush().map_err(crate::Error::Io)
+    }
+
+    /// Writes a collection of sentences on the current stream.
+    pub fn write_sentences<T: Sentence>(&mut self, sentences: &[T]) -> crate::Result<()> {
+        for sentence in sentences {
+            let encoded = sentence.encode();
+            let joined = join_sentence(encoded);
+            self.write_literal(&joined)?;
+        }
+        self.flush().map_err(crate::Error::Io)
+    }
+
+    /// Writes a literal string to the current stream.
+    /// Note: the literal string MUST end with a new-line character (`\n`).
+    ///
+    /// Note: does not automatically flush.
+    pub fn write_literal(&mut self, literal: &str) -> crate::Result<()> {
+        assert!(literal.ends_with('\n'));
+        self.write_all(literal.as_bytes())?;
+        Ok(())
+    }
+
+    /// Reads a literal string from the current stream.
+    /// Note: the literal string will ends with a new-line character (`\n`).
+    pub fn read_literal(reader: &mut BufReader<&mut Self>) -> crate::Result<String> {
+        let mut raw = String::new();
+        reader.read_line(&mut raw)?;
+        Ok(raw)
+    }
+
+    /// Reads a sentence from the given `BufReader`.
+    pub fn read_sentence<T: Sentence>(reader: &mut BufReader<&mut Self>) -> crate::Result<T> {
+        let raw = Self::read_literal(reader)?;
+        if raw.is_empty() {
+            return Err(Error::eof());
+        }
+        let split = split_sentence(raw).ok_or(crate::NutError::NotProcessable)?;
+        T::decode(split)
+            .ok_or(Error::Nut(NutError::InvalidArgument))?
+            .into()
+    }
+
+    /// Reads all sentences in the buffer until the given `matcher` function evaluates to `true`.
+    ///
+    /// The final sentence is excluded.
+    pub fn read_sentences_until<T: Sentence, F: Fn(&T) -> bool>(
+        reader: &mut BufReader<&mut Self>,
+        matcher: F,
+    ) -> crate::Result<Vec<T>> {
+        let mut result = Vec::new();
+        let max_iter = 1000; // Exit after 1000 lines to prevent overflow.
+        for _ in 0..max_iter {
+            let sentence: T = Self::read_sentence(reader)?;
+            if matcher(&sentence) {
+                return Ok(result);
+            } else {
+                result.push(sentence);
+            }
+        }
+        Err(Error::Io(std::io::Error::new(
+            std::io::ErrorKind::Interrupted,
+            "Reached maximum read capacity.",
+        )))
+    }
+
+    /// Initializes a new `BufReader` for the current stream.
+    pub fn buffer(&mut self) -> BufReader<&mut Self> {
+        BufReader::new(self)
+    }
 }
 
 impl Read for ConnectionStream {
@@ -47,6 +130,8 @@ impl Read for ConnectionStream {
             Self::SslClient(stream) => stream.read(buf),
             #[cfg(feature = "ssl")]
             Self::SslServer(stream) => stream.read(buf),
+            #[cfg(test)]
+            Self::Mock(stream) => stream.read(buf),
         }
     }
 }
@@ -59,6 +144,12 @@ impl Write for ConnectionStream {
             Self::SslClient(stream) => stream.write(buf),
             #[cfg(feature = "ssl")]
             Self::SslServer(stream) => stream.write(buf),
+            #[cfg(test)]
+            Self::Mock(stream) => {
+                let len = buf.len();
+                stream.push_bytes_to_read(buf);
+                Ok(len)
+            }
         }
     }
 
@@ -69,6 +160,113 @@ impl Write for ConnectionStream {
             Self::SslClient(stream) => stream.flush(),
             #[cfg(feature = "ssl")]
             Self::SslServer(stream) => stream.flush(),
+            #[cfg(test)]
+            Self::Mock(stream) => stream.flush(),
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::ConnectionStream;
+    use crate::proto::{ClientSentences, ServerSentences};
+    use std::io::{Read, Write};
+
+    #[test]
+    fn read_write_sentence() {
+        let mut client_stream = mockstream::SharedMockStream::new();
+        let mut server_stream = client_stream.clone();
+
+        let mut client_stream = ConnectionStream::Mock(client_stream);
+        let mut server_stream = ConnectionStream::Mock(server_stream);
+
+        // Client requests list of UPS devices
+        client_stream
+            .write_sentence(&ServerSentences::QueryListUps {})
+            .expect("Failed to write LIST UPS");
+
+        // Server reads query for list of UPS devices
+        let mut server_buffer = server_stream.buffer();
+        let sentence: ServerSentences =
+            ConnectionStream::read_sentence(&mut server_buffer).expect("Failed to read LIST UPS");
+        assert_eq!(sentence, ServerSentences::QueryListUps {});
+
+        // Server sends list of UPS devices.
+        server_stream
+            .write_sentences(&[
+                ClientSentences::BeginListUps {},
+                ClientSentences::RespondUps {
+                    ups_name: "nutdev0".into(),
+                    description: "A NUT device.".into(),
+                },
+                ClientSentences::RespondUps {
+                    ups_name: "nutdev1".into(),
+                    description: "Another NUT device.".into(),
+                },
+                ClientSentences::EndListUps {},
+            ])
+            .expect("Failed to write UPS LIST");
+
+        // Client reads list of UPS devices.
+        let mut client_buffer = client_stream.buffer();
+        let sentence: ClientSentences = ConnectionStream::read_sentence(&mut client_buffer)
+            .expect("Failed to read BEGIN LIST UPS");
+        assert_eq!(sentence, ClientSentences::BeginListUps {});
+
+        let sentences: Vec<ClientSentences> =
+            ConnectionStream::read_sentences_until(&mut client_buffer, |s| {
+                matches!(s, ClientSentences::EndListUps {})
+            })
+            .expect("Failed to read UPS items");
+
+        assert_eq!(
+            sentences,
+            vec![
+                ClientSentences::RespondUps {
+                    ups_name: "nutdev0".into(),
+                    description: "A NUT device.".into(),
+                },
+                ClientSentences::RespondUps {
+                    ups_name: "nutdev1".into(),
+                    description: "Another NUT device.".into(),
+                },
+            ]
+        );
+
+        // Client sends login
+        client_stream
+            .write_sentence(&ServerSentences::ExecLogin {
+                ups_name: "nutdev0".into(),
+            })
+            .expect("Failed to write LOGIN nutdev0");
+
+        // Server receives login
+        let mut server_buffer = server_stream.buffer();
+        let sentence: ServerSentences = ConnectionStream::read_sentence(&mut server_buffer)
+            .expect("Failed to read LOGIN nutdev0");
+        assert_eq!(
+            sentence,
+            ServerSentences::ExecLogin {
+                ups_name: "nutdev0".into()
+            }
+        );
+
+        // Server rejects login
+        server_stream
+            .write_sentence(&ClientSentences::RespondErr {
+                message: "USERNAME-REQUIRED".into(),
+                extras: vec![],
+            })
+            .expect("Failed to write ERR USERNAME-REQUIRED");
+
+        // Client expects error
+        let mut client_buffer = client_stream.buffer();
+        let error: crate::Error =
+            ConnectionStream::read_sentence::<ClientSentences>(&mut client_buffer)
+                .expect_err("Failed to read ERR");
+        assert!(matches!(
+            error,
+            crate::Error::Nut(crate::NutError::UsernameRequired)
+        ));
+    }
+}

rups/src/error.rs

@@ -62,6 +62,8 @@ pub enum NutError {
     SslInvalidHostname,
     /// Occurs when the client used a feature that is disabled by the server.
     FeatureNotConfigured,
+    /// The client or server sent a message that could not be processed.
+    NotProcessable,
     /// Generic (usually internal) client error.
     Generic(String),
 }
@@ -99,14 +101,15 @@ impl fmt::Display for NutError {
                 "Given hostname cannot be used for a strict SSL connection"
             ),
             Self::FeatureNotConfigured => write!(f, "Feature not configured by server"),
+            Self::NotProcessable => write!(f, "Message could not be processed"),
             Self::Generic(msg) => write!(f, "NUT error: {}", msg),
         }
     }
 }
 
-impl<T: AsRef<ClientSentences>> From<T> for NutError {
-    fn from(sentence: T) -> Self {
-        if let ClientSentences::RespondErr { message, .. } = sentence.as_ref() {
+impl From<ClientSentences> for NutError {
+    fn from(sentence: ClientSentences) -> Self {
+        if let ClientSentences::RespondErr { message, .. } = sentence {
             match message.as_str() {
                 "ACCESS-DENIED" => Self::AccessDenied,
                 "UNKNOWN-UPS" => Self::UnknownUps,
@@ -135,7 +138,7 @@ impl<T: AsRef<ClientSentences>> From<T> for NutError {
             }
         } else {
             // This is not supposed to happen...
-            panic!("Cannot convert {:?} into NutError", sentence.as_ref());
+            panic!("Cannot convert {:?} into NutError", sentence);
         }
     }
 }
@@ -163,6 +166,14 @@ impl Error {
     pub fn generic<T: ToString>(message: T) -> Self {
         NutError::generic(message.to_string()).into()
     }
+
+    /// Constructs an EOF (end-of-file) error.
+    pub fn eof() -> Self {
+        Self::Io(io::Error::new(
+            io::ErrorKind::UnexpectedEof,
+            "Reached end of stream while sentence was expected",
+        ))
+    }
 }
 
 impl fmt::Display for Error {

rups/src/proto/client.rs

@@ -1,4 +1,5 @@
 use crate::proto::impl_sentences;
+use crate::{Error, NutError};
 
 impl_sentences! {
     /// A generic successful response with no additional data.
@@ -456,9 +457,20 @@ impl_sentences! {
     ),
 }
 
+#[allow(clippy::from_over_into)]
+impl Into<crate::Result<Self>> for Sentences {
+    fn into(self) -> crate::Result<Sentences> {
+        if let Sentences::RespondErr { .. } = &self {
+            Err(Error::Nut(NutError::from(self)))
+        } else {
+            Ok(self)
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
-    use crate::proto::test_encode_decode;
+    use crate::proto::{test_encode_decode, Sentence};
 
     use super::Sentences;
 

rups/src/proto/mod.rs

@@ -104,6 +104,15 @@ macro_rules! impl_words {
     };
 }
 
+/// A NUT protocol sentence that can be encoded and decoded from a Vector of strings.
+pub trait Sentence: Sized + Into<crate::Result<Self>> {
+    /// Decodes a sentence. Returns `None` if the pattern cannot be recognized.
+    fn decode(raw: Vec<String>) -> Option<Self>;
+
+    /// Encodes the sentence.
+    fn encode(&self) -> Vec<&str>;
+}
+
 /// Implements the list of sentences, which are combinations
 /// of words that form commands (serverbound) and responses (clientbound).
 macro_rules! impl_sentences {
@@ -147,9 +156,8 @@ macro_rules! impl_sentences {
             )*
         }
 
-        impl Sentences {
-            /// Decodes a sentence. Returns `None` if the pattern cannot be recognized.
-            pub(crate) fn decode(raw: Vec<String>) -> Option<Sentences> {
+        impl crate::proto::Sentence for Sentences {
+            fn decode(raw: Vec<String>) -> Option<Sentences> {
                 use super::{Word::*, *};
                 use Sentences::*;
                 let words = Word::decode_words(raw.as_slice());
@@ -168,8 +176,7 @@ macro_rules! impl_sentences {
                 None
             }
 
-            /// Encodes the sentence.
-            pub(crate) fn encode(&self) -> Vec<&str> {
+            fn encode(&self) -> Vec<&str> {
                 use super::Word::*;
                 match self {
                     $(

rups/src/proto/server.rs

@@ -91,6 +91,15 @@ impl_sentences! {
             3: cmd_name,
         }
     ),
+    /// Client requests the list of UPS devices.
+    QueryListUps (
+        {
+            0: List,
+            1: Ups,
+            2: EOL,
+        },
+        {}
+    ),
     /// Client requests the list of variables for the given `ups_name` device.
     QueryListVar (
         {
@@ -313,10 +322,17 @@ impl_sentences! {
     ),
 }
 
+#[allow(clippy::from_over_into)]
+impl Into<crate::Result<Self>> for Sentences {
+    fn into(self) -> crate::Result<Sentences> {
+        Ok(self)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::Sentences;
-    use crate::proto::test_encode_decode;
+    use crate::proto::{test_encode_decode, Sentence};
     #[test]
     fn test_encode_decode() {
         test_encode_decode!(
@@ -371,6 +387,10 @@ mod tests {
                 ups_name: "nutdev".into(),
             }
         );
+        test_encode_decode!(
+            ["LIST", "UPS"] <=>
+            Sentences::QueryListUps {}
+        );
         test_encode_decode!(
             ["LIST", "RW", "nutdev"] <=>
             Sentences::QueryListRw {

rups/src/proto/util.rs

@@ -3,7 +3,7 @@
 ///
 /// Returns `None` if the sentence cannot be split safely (usually unbalanced quotation marks).
 pub fn split_sentence<T: AsRef<str>>(sentence: T) -> Option<Vec<String>> {
-    shell_words::split(sentence.as_ref()).ok()
+    shell_words::split(sentence.as_ref().trim_end_matches('\n')).ok()
 }
 
 /// Joins a collection of words (`&str`) into one sentence string,
@@ -13,7 +13,7 @@ where
     I: IntoIterator<Item = S>,
     S: AsRef<str>,
 {
-    shell_words::join(words)
+    format!("{}\n", shell_words::join(words))
 }
 
 #[cfg(test)]
@@ -34,8 +34,8 @@ mod tests {
 
     #[test]
     fn test_join() {
-        assert_eq!(join_sentence(vec!["AbC", "dEf", "GHi"]), "AbC dEf GHi",);
-        assert_eq!(join_sentence(vec!["AbC dEf", "GHi"]), "'AbC dEf' GHi",);
-        assert_eq!(join_sentence(vec!["\"AbC dEf", "GHi"]), "'\"AbC dEf' GHi",);
+        assert_eq!(join_sentence(vec!["AbC", "dEf", "GHi"]), "AbC dEf GHi\n",);
+        assert_eq!(join_sentence(vec!["AbC dEf", "GHi"]), "'AbC dEf' GHi\n",);
+        assert_eq!(join_sentence(vec!["\"AbC dEf", "GHi"]), "'\"AbC dEf' GHi\n",);
     }
 }