Async support with tokio

Author: dani-garcia

Cargo.toml

@@ -1,26 +1,21 @@
 [package]
-name = "ftp"
-version = "3.0.1"
-authors = ["Matt McCoy <[email protected]>"]
-documentation = "https://docs.rs/ftp/"
-repository = "https://github.com/mattnenterprise/rust-ftp"
+name = "async_ftp"
+version = "4.0.0"
+authors = ["Daniel García <[email protected]>", "Matt McCoy <[email protected]>"]
+documentation = "https://docs.rs/async_ftp/"
+repository = "https://github.com/dani-garcia/rust_async_ftp"
 description = "FTP client for Rust"
 readme = "README.md"
 license = "Apache-2.0/MIT"
 edition = "2018"
 keywords = ["ftp"]
 categories = ["network-programming"]
 
-[badges]
-travis-ci = { repository = "mattnenterprise/rust-ftp" }
-
-[lib]
-name ="ftp"
-path = "src/lib.rs"
-
 [features]
+# default = ["secure"]
+
 # Enable support of FTPS which requires openssl
-secure = ["openssl"]
+secure = ["tokio-rustls"]
 
 # Add debug output (to STDOUT) of commands sent to the server
 # and lines read from the server
@@ -31,6 +26,6 @@ lazy_static = "1.4.0"
 regex = "1.3.9"
 chrono = "0.4.11"
 
-[dependencies.openssl]
-version = "0.10.29"
-optional = true
+tokio = { version = "0.2.21", features = ["tcp", "dns", "io-util"] }
+tokio-rustls = { version = "0.13.1", optional = true }
+pin-project = "0.4.17"

README.md

@@ -21,36 +21,43 @@ ftp = { version = "<version>", features = ["secure"] }
 
 ## Usage
 ```rust
-extern crate ftp;
-
 use std::str;
 use std::io::Cursor;
-use ftp::FtpStream;
+use async_ftp::FtpStream;
 
-fn main() {
+async fn async_main() -> Result<(), Box<dyn std::error::Error>> {
     // Create a connection to an FTP server and authenticate to it.
-    let mut ftp_stream = FtpStream::connect("127.0.0.1:21").unwrap();
-    let _ = ftp_stream.login("username", "password").unwrap();
+    let mut ftp_stream = FtpStream::connect("172.25.82.139:21").await?;
+    let _ = ftp_stream.login("username", "password").await?;
 
     // Get the current directory that the client will be reading from and writing to.
-    println!("Current directory: {}", ftp_stream.pwd().unwrap());
+    println!("Current directory: {}", ftp_stream.pwd().await?);
 
     // Change into a new directory, relative to the one we are currently in.
-    let _ = ftp_stream.cwd("test_data").unwrap();
+    let _ = ftp_stream.cwd("test_data").await?;
 
     // Retrieve (GET) a file from the FTP server in the current working directory.
-    let remote_file = ftp_stream.simple_retr("ftpext-charter.txt").unwrap();
-    println!("Read file with contents\n{}\n", str::from_utf8(&remote_file.into_inner()).unwrap());
+    let remote_file = ftp_stream.simple_retr("ftpext-charter.txt").await?;
+    println!("Read file with contents\n{}\n", str::from_utf8(&remote_file.into_inner()).await?);
 
     // Store (PUT) a file from the client to the current working directory of the server.
     let mut reader = Cursor::new("Hello from the Rust \"ftp\" crate!".as_bytes());
-    let _ = ftp_stream.put("greeting.txt", &mut reader);
+    let _ = ftp_stream.put("greeting.txt", &mut reader).await?;
     println!("Successfully wrote greeting.txt");
 
     // Terminate the connection to the server.
     let _ = ftp_stream.quit();
 }
 
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tokio::runtime::Builder::new()
+        .threaded_scheduler()
+        .enable_all()
+        .build()
+        .unwrap()
+        .block_on(async_main())
+}
+
 ```
 
 ## License

examples/connecting.rs

@@ -1,31 +1,37 @@
-extern crate ftp;
-
-use std::str;
+use async_ftp::{FtpError, FtpStream};
 use std::io::Cursor;
-use ftp::{FtpStream, FtpError};
+use std::str;
 
-fn test_ftp(addr: &str, user: &str, pass: &str) -> Result<(), FtpError> {
-    let mut ftp_stream = FtpStream::connect((addr, 21)).unwrap();
-    ftp_stream.login(user, pass).unwrap();
-    println!("current dir: {}", ftp_stream.pwd().unwrap());
+async fn test_ftp(addr: &str, user: &str, pass: &str) -> Result<(), FtpError> {
+    let mut ftp_stream = FtpStream::connect((addr, 21)).await?;
+    ftp_stream.login(user, pass).await?;
+    println!("current dir: {}", ftp_stream.pwd().await?);
 
-    ftp_stream.cwd("test_data").unwrap();
+    ftp_stream.cwd("test_data").await?;
 
     // An easy way to retrieve a file
-    let cursor = ftp_stream.simple_retr("ftpext-charter.txt").unwrap();
+    let cursor = ftp_stream.simple_retr("ftpext-charter.txt").await?;
     let vec = cursor.into_inner();
     let text = str::from_utf8(&vec).unwrap();
     println!("got data: {}", text);
 
     // Store a file
     let file_data = format!("Some awesome file data man!!");
     let mut reader = Cursor::new(file_data.into_bytes());
-    ftp_stream.put("my_random_file.txt", &mut reader).unwrap();
+    ftp_stream.put("my_random_file.txt", &mut reader).await?;
 
-    ftp_stream.quit()
+    ftp_stream.quit().await
 }
 
 fn main() {
-    test_ftp("127.0.0.1", "anonymous", "[email protected]").unwrap();
+    let future = test_ftp("172.25.82.139", "anonymous", "[email protected]");
+
+    tokio::runtime::Builder::new()
+        .enable_all()
+        .build()
+        .unwrap()
+        .block_on(future)
+        .unwrap();
+
     println!("test successful")
 }

src/data_stream.rs

@@ -1,15 +1,17 @@
-use std::io::{Read, Write, Result};
-use std::net::TcpStream;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::net::TcpStream;
 #[cfg(feature = "secure")]
-use openssl::ssl::SslStream;
-
+use tokio_rustls::client::TlsStream;
 
 /// Data Stream used for communications
-#[derive(Debug)]
+#[pin_project::pin_project(project = DataStreamProj)]
 pub enum DataStream {
-    Tcp(TcpStream),
+    Tcp(#[pin] TcpStream),
     #[cfg(feature = "secure")]
-    Ssl(SslStream<TcpStream>),
+    Ssl(#[pin] TlsStream<TcpStream>),
 }
 
 impl DataStream {
@@ -18,54 +20,69 @@ impl DataStream {
         match self {
             DataStream::Tcp(stream) => stream,
             #[cfg(feature = "secure")]
-            DataStream::Ssl(stream) => stream.get_ref().try_clone().unwrap(),
+            DataStream::Ssl(stream) => stream.into_inner().0,
         }
     }
 
     /// Test if the stream is secured
     pub fn is_ssl(&self) -> bool {
         match self {
             #[cfg(feature = "secure")]
-            &DataStream::Ssl(_) => true,
-            _ => false
+            DataStream::Ssl(_) => true,
+            _ => false,
         }
     }
 
     /// Returns a reference to the underlying TcpStream.
     pub fn get_ref(&self) -> &TcpStream {
         match self {
-            &DataStream::Tcp(ref stream) => stream,
+            DataStream::Tcp(ref stream) => stream,
             #[cfg(feature = "secure")]
-            &DataStream::Ssl(ref stream) => stream.get_ref(),
+            DataStream::Ssl(ref stream) => stream.get_ref().0,
         }
     }
 }
 
-impl Read for DataStream {
-    fn read(&mut self, buf: &mut [u8]) -> Result<usize> {
-        match self {
-            &mut DataStream::Tcp(ref mut stream) => stream.read(buf),
+impl AsyncRead for DataStream {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        match self.project() {
+            DataStreamProj::Tcp(stream) => stream.poll_read(cx, buf),
             #[cfg(feature = "secure")]
-            &mut DataStream::Ssl(ref mut stream) => stream.read(buf),
+            DataStreamProj::Ssl(stream) => stream.poll_read(cx, buf),
         }
     }
 }
 
+impl AsyncWrite for DataStream {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        match self.project() {
+            DataStreamProj::Tcp(stream) => stream.poll_write(cx, buf),
+            #[cfg(feature = "secure")]
+            DataStreamProj::Ssl(stream) => stream.poll_write(cx, buf),
+        }
+    }
 
-impl Write for DataStream {
-    fn write(&mut self, buf: &[u8]) -> Result<usize> {
-        match self {
-            &mut DataStream::Tcp(ref mut stream) => stream.write(buf),
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        match self.project() {
+            DataStreamProj::Tcp(stream) => stream.poll_flush(cx),
             #[cfg(feature = "secure")]
-            &mut DataStream::Ssl(ref mut stream) => stream.write(buf),
+            DataStreamProj::Ssl(stream) => stream.poll_flush(cx),
         }
     }
 
-    fn flush(&mut self) -> Result<()> {
-        match self {
-            &mut DataStream::Tcp(ref mut stream) => stream.flush(),
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        match self.project() {
+            DataStreamProj::Tcp(stream) => stream.poll_shutdown(cx),
             #[cfg(feature = "secure")]
-            &mut DataStream::Ssl(ref mut stream) => stream.flush(),
+            DataStreamProj::Ssl(stream) => stream.poll_shutdown(cx),
         }
     }
 }