send_with_timeout: apply timeout to message sending

Previously all the commands, such as RCPT and DATA, were using custom
user-supplied timeout, but the most expensive operation of sending the
message data and waiting for before-queue content filters to finish were
using default SMTP timeout.

This commit also removes the timeout from message reading operation. It
normally happens locally and otherwise I/O operations have their own
timeouts, e.g.  if the message is read from NFS mount.

Author: Alexander Krotov

src/smtp/client/inner.rs

@@ -171,29 +171,35 @@ impl<S: Connector + Write + Read + Unpin> InnerClient<S> {
     }
 
     /// Sends the message content.
-    pub async fn message<T: Read + Unpin>(mut self: Pin<&mut Self>, message: T) -> SmtpResult {
-        let timeout = self.timeout;
-        with_timeout(timeout.as_ref(), async {
-            let mut codec = ClientCodec::new();
-            let mut message_reader = BufReader::new(message);
+    pub(crate) async fn message_with_timeout<T: Read + Unpin>(
+        mut self: Pin<&mut Self>,
+        message: T,
+        timeout: Option<&Duration>,
+    ) -> SmtpResult {
+        let mut codec = ClientCodec::new();
 
-            let mut message_bytes = Vec::new();
-            message_reader.read_to_end(&mut message_bytes).await?;
+        let mut message_reader = BufReader::new(message);
 
-            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 message_bytes = Vec::new();
+        message_reader.read_to_end(&mut message_bytes).await?;
+
+        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)?;
+        let mut stream = this.stream.as_pin_mut().ok_or(Error::NoStream)?;
 
+        let res: Result<(), Error> = with_timeout(timeout, async {
             codec.encode(&message_bytes, &mut stream).await?;
             stream.write_all(b"\r\n.\r\n").await?;
-
-            self.read_response_no_timeout().await
+            Ok(())
         })
-        .await
+        .await;
+        res?;
+
+        with_timeout(timeout, self.read_response_no_timeout()).await
     }
 
     /// Send the given SMTP command to the server.

src/smtp/smtp_client.rs

@@ -552,7 +552,10 @@ impl<'a> Transport<'a> for SmtpTransport {
         // Data
         try_smtp!(client.as_mut().command(DataCommand).await, self);
 
-        let res = client.as_mut().message(email.message()).await;
+        let res = client
+            .as_mut()
+            .message_with_timeout(email.message(), timeout)
+            .await;
 
         // Message content
         if let Ok(result) = &res {