Merge pull request #2 from dodomorandi/multiplatform_lines

Split lines according to both dos and unix conventions

Author: dani-garcia

Cargo.toml

@@ -29,3 +29,7 @@ chrono = "0.4.11"
 tokio = { version = "0.2.21", features = ["tcp", "dns", "io-util"] }
 tokio-rustls = { version = "0.13.1", optional = true }
 pin-project = "0.4.17"
+
+[dev-dependencies.tokio]
+version = "0.2.21"
+features = [ "macros", "stream", "io-util" ]

src/ftp.rs

@@ -9,7 +9,8 @@ use chrono::{DateTime, Utc};
 use regex::Regex;
 
 use tokio::io::{
-    copy, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter,
+    copy, AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWriteExt, BufReader,
+    BufWriter,
 };
 use tokio::net::{TcpStream, ToSocketAddrs};
 
@@ -435,28 +436,38 @@ impl FtpStream {
         open_code: u32,
         close_code: &[u32],
     ) -> Result<Vec<String>> {
-        let mut lines: Vec<String> = Vec::new();
-
-        let mut data_stream = BufReader::new(self.data_command(&cmd).await?);
+        let data_stream = BufReader::new(self.data_command(&cmd).await?);
         self.read_response_in(&[open_code, status::ALREADY_OPEN])
             .await?;
+        let lines = Self::get_lines_from_stream(data_stream).await?;
+        self.read_response_in(close_code).await?;
+        Ok(lines)
+    }
 
-        let mut line = String::new();
+    /// Consume a stream and return a vector of lines
+    async fn get_lines_from_stream<R>(data_stream: R) -> Result<Vec<String>>
+    where
+        R: AsyncBufRead + Unpin,
+    {
+        let mut lines: Vec<String> = Vec::new();
+
+        let mut lines_stream = data_stream.lines();
         loop {
-            match data_stream.read_to_string(&mut line).await {
-                Ok(0) => break,
-                Ok(_) => lines.extend(
-                    line.split("\r\n")
-                        .map(String::from)
-                        .filter(|s| !s.is_empty()),
-                ),
-                Err(err) => return Err(FtpError::ConnectionError(err)),
-            };
-        }
-        drop(data_stream);
+            let line = lines_stream
+                .next_line()
+                .await
+                .map_err(FtpError::ConnectionError)?;
 
-        self.read_response_in(close_code).await?;
-        Ok(lines)
+            match line {
+                Some(line) => {
+                    if line.is_empty() {
+                        continue;
+                    }
+                    lines.push(line);
+                }
+                None => break Ok(lines),
+            }
+        }
     }
 
     /// Execute `LIST` command which returns the detailed file listing in human readable format.
@@ -597,3 +608,37 @@ impl FtpStream {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::FtpStream;
+    use tokio::{io::stream_reader, stream};
+
+    #[tokio::test]
+    async fn list_command_dos_newlines() {
+        let data_stream = stream_reader(stream::once(Ok(
+            b"Hello\r\nWorld\r\n\r\nBe\r\nHappy\r\n" as &[u8]
+        )));
+
+        assert_eq!(
+            FtpStream::get_lines_from_stream(data_stream).await.unwrap(),
+            ["Hello", "World", "Be", "Happy"]
+                .iter()
+                .map(<&str>::to_string)
+                .collect::<Vec<_>>()
+        );
+    }
+
+    #[tokio::test]
+    async fn list_command_unix_newlines() {
+        let data_stream = stream_reader(stream::once(Ok(b"Hello\nWorld\n\nBe\nHappy\n" as &[u8])));
+
+        assert_eq!(
+            FtpStream::get_lines_from_stream(data_stream).await.unwrap(),
+            ["Hello", "World", "Be", "Happy"]
+                .iter()
+                .map(<&str>::to_string)
+                .collect::<Vec<_>>()
+        );
+    }
+}