First try

Author: Hocuri

src/file/mod.rs

@@ -3,7 +3,7 @@
 //! It can be useful for testing purposes, or if you want to keep track of sent messages.
 //!
 
-use std::path::PathBuf;
+use std::{path::PathBuf, time::Duration};
 
 use async_std::fs::File;
 use async_std::path::Path;
@@ -71,4 +71,12 @@ impl<'a> Transport<'a> for FileTransport {
             .await?;
         Ok(())
     }
+
+    async fn send_with_timeout(
+        &mut self,
+        email: SendableEmail,
+        _timeout: Option<&Duration>,
+    ) -> Self::Result {
+        self.send(email).await // Writing to a file does not have a timeout, so just ignore it.
+    }
 }

src/lib.rs

@@ -34,6 +34,7 @@ pub use crate::smtp::client::net::ClientTlsParameters;
 pub use crate::smtp::{ClientSecurity, SmtpClient, SmtpTransport};
 
 use async_trait::async_trait;
+use std::time::Duration;
 
 /// Transport method for emails
 #[async_trait]
@@ -43,4 +44,10 @@ pub trait Transport<'a> {
 
     /// Sends the email
     async fn send(&mut self, email: SendableEmail) -> Self::Result;
+
+    async fn send_with_timeout(
+        &mut self,
+        email: SendableEmail,
+        timeout: Option<&Duration>,
+    ) -> Self::Result;
 }

src/sendmail/mod.rs

@@ -10,7 +10,10 @@ use async_trait::async_trait;
 use log::info;
 use std::convert::AsRef;
 use std::io::prelude::*;
-use std::process::{Command, Stdio};
+use std::{
+    process::{Command, Stdio},
+    time::Duration,
+};
 
 pub mod error;
 
@@ -84,9 +87,17 @@ impl<'a> Transport<'a> for SendmailTransport {
         .await?;
 
         if output.status.success() {
-            return Ok(());
+            Ok(())
+        } else {
+            Err(error::Error::Client(String::from_utf8(output.stderr)?))
         }
+    }
 
-        Err(error::Error::Client(String::from_utf8(output.stderr)?))
+    async fn send_with_timeout(
+        &mut self,
+        email: SendableEmail,
+        _timeout: Option<&Duration>,
+    ) -> Self::Result {
+        self.send(email).await
     }
 }

src/smtp/client/inner.rs

@@ -99,7 +99,7 @@ impl<S: Connector + Write + Read + Unpin> InnerClient<S> {
     }
 
     /// Get the read and write timeout.
-    pub fn timeout(&mut self) -> Option<&Duration> {
+    pub fn timeout(&self) -> Option<&Duration> {
         self.timeout.as_ref()
     }
 
@@ -194,25 +194,41 @@ impl<S: Connector + Write + Read + Unpin> InnerClient<S> {
         })
         .await?;
 
-        self.read_response().await
+        let timeout = self.timeout().cloned();
+        self.read_response(timeout.as_ref()).await
     }
 
     /// Send the given SMTP command to the server.
-    pub async fn command<C: Display>(mut self: Pin<&mut Self>, command: C) -> SmtpResult {
-        self.as_mut().write(command.to_string().as_bytes()).await?;
-        self.read_response().await
+    pub async fn command<C: Display>(self: Pin<&mut Self>, command: C) -> SmtpResult {
+        let timeout = self.timeout.clone();
+        self.command_with_timeout(command, timeout.as_ref()).await
+    }
+
+    pub async fn command_with_timeout<C: Display>(
+        mut self: Pin<&mut Self>,
+        command: C,
+        timeout: Option<&Duration>,
+    ) -> SmtpResult {
+        self.as_mut()
+            .write(command.to_string().as_bytes(), timeout)
+            .await?;
+        self.read_response(timeout).await
     }
 
     /// Writes the given data to the server.
-    async fn write(mut self: Pin<&mut Self>, string: &[u8]) -> Result<(), Error> {
+    async fn write(
+        mut self: Pin<&mut Self>,
+        string: &[u8],
+        timeout: Option<&Duration>,
+    ) -> Result<(), Error> {
         if self.stream.is_none() {
             return Err(From::from("Connection closed"));
         }
         let this = self.as_mut().project();
         let _: Pin<&mut Option<S>> = this.stream;
         let mut stream = this.stream.as_pin_mut().ok_or(Error::NoStream)?;
 
-        with_timeout(this.timeout.as_ref(), async move {
+        with_timeout(timeout, async move {
             stream.write_all(string).await?;
             stream.flush().await?;
             Ok(())
@@ -227,15 +243,15 @@ impl<S: Connector + Write + Read + Unpin> InnerClient<S> {
     }
 
     /// Read an SMTP response from the wire.
-    pub async fn read_response(mut self: Pin<&mut Self>) -> SmtpResult {
+    pub async fn read_response(mut self: Pin<&mut Self>, timeout: Option<&Duration>) -> SmtpResult {
         let this = self.as_mut().project();
         let stream = this.stream.as_pin_mut().ok_or(Error::NoStream)?;
 
         let mut reader = BufReader::new(stream);
         let mut buffer = String::with_capacity(100);
 
         loop {
-            let read = with_timeout(this.timeout.as_ref(), reader.read_line(&mut buffer)).await?;
+            let read = with_timeout(timeout, reader.read_line(&mut buffer)).await?;
             if read == 0 {
                 break;
             }

src/smtp/smtp_client.rs

@@ -306,7 +306,8 @@ impl<'a> SmtpTransport {
                 .await?;
 
             client.set_timeout(self.client_info.timeout);
-            let _response = client.read_response().await?;
+            let timeout = client.timeout().cloned();
+            let _response = client.read_response(timeout.as_ref()).await?;
         }
 
         self.post_connect().await
@@ -498,6 +499,15 @@ impl<'a> Transport<'a> for SmtpTransport {
 
     /// Sends an email
     async fn send(&mut self, email: SendableEmail) -> SmtpResult {
+        let timeout = self.client.timeout().cloned();
+        self.send_with_timeout(email, timeout.as_ref()).await
+    }
+
+    async fn send_with_timeout(
+        &mut self,
+        email: SendableEmail,
+        timeout: Option<&Duration>,
+    ) -> Self::Result {
         let message_id = email.message_id().to_string();
 
         // Mail
@@ -516,10 +526,10 @@ impl<'a> Transport<'a> for SmtpTransport {
         try_smtp!(
             client
                 .as_mut()
-                .command(MailCommand::new(
-                    email.envelope().from().cloned(),
-                    mail_options,
-                ))
+                .command_with_timeout(
+                    MailCommand::new(email.envelope().from().cloned(), mail_options),
+                    timeout
+                )
                 .await,
             self
         );
@@ -529,7 +539,7 @@ impl<'a> Transport<'a> for SmtpTransport {
             try_smtp!(
                 client
                     .as_mut()
-                    .command(RcptCommand::new(to_address.clone(), vec![]))
+                    .command_with_timeout(RcptCommand::new(to_address.clone(), vec![]), timeout)
                     .await,
                 self
             );

src/stub/mod.rs

@@ -7,6 +7,7 @@ use log::info;
 
 use crate::SendableEmail;
 use crate::Transport;
+use std::time::Duration;
 
 /// This transport logs the message envelope and returns the given response
 #[derive(Debug, Clone, Copy)]
@@ -45,4 +46,12 @@ impl<'a> Transport<'a> for StubTransport {
         );
         self.response
     }
+    async fn send_with_timeout(
+        &mut self,
+        email: SendableEmail,
+        timeout: Option<&Duration>,
+    ) -> Self::Result {
+        info!("Timeout: {:?}", timeout);
+        self.send(email).await
+    }
 }