wip

src/message/body.rs

@@ -1,11 +1,11 @@
-use std::{mem, ops::Deref};
+use std::{mem, ops::Deref, sync::Arc};
 
 use crate::message::header::ContentTransferEncoding;
 
 /// A [`Message`][super::Message] or [`SinglePart`][super::SinglePart] body that has already been encoded.
 #[derive(Debug, Clone)]
 pub struct Body {
-    buf: Vec<u8>,
+    buf: Arc<[u8]>,
     encoding: ContentTransferEncoding,
 }
 
@@ -39,7 +39,7 @@ impl Body {
 
         let encoding = buf.encoding(false);
         buf.encode_crlf();
-        Self::new_impl(buf.into(), encoding)
+        Self::new_impl(Vec::from(buf).into(), encoding)
     }
 
     /// Encode the supplied `buf`, using the provided `encoding`.
@@ -77,7 +77,7 @@ impl Body {
         }
 
         buf.encode_crlf();
-        Ok(Self::new_impl(buf.into(), encoding))
+        Ok(Self::new_impl(Vec::from(buf).into(), encoding))
     }
 
     /// Builds a new `Body` using a pre-encoded buffer.
@@ -87,11 +87,14 @@ impl Body {
     /// `buf` shouldn't contain non-ascii characters, lines longer than 1000 characters or nul bytes.
     #[inline]
     pub fn dangerous_pre_encoded(buf: Vec<u8>, encoding: ContentTransferEncoding) -> Self {
-        Self { buf, encoding }
+        Self {
+            buf: buf.into(),
+            encoding,
+        }
     }
 
     /// Encodes the supplied `buf` using the provided `encoding`
-    fn new_impl(buf: Vec<u8>, encoding: ContentTransferEncoding) -> Self {
+    fn new_impl(buf: Arc<[u8]>, encoding: ContentTransferEncoding) -> Self {
         match encoding {
             ContentTransferEncoding::SevenBit
             | ContentTransferEncoding::EightBit
@@ -133,7 +136,16 @@ impl Body {
 
     /// Consumes `Body` and returns the inner `Vec<u8>`
     #[inline]
+    #[deprecated(
+        note = "The inner memory is not stored into `Vec<u8>` anymore. Consider using `into_inner`"
+    )]
     pub fn into_vec(self) -> Vec<u8> {
+        self.buf.to_vec()
+    }
+
+    /// Consumes `Body` and returns the inner `Arc<[u8]>`
+    #[inline]
+    pub fn into_inner(self) -> Arc<[u8]> {
         self.buf
     }
 }

src/message/header/mod.rs

@@ -18,6 +18,7 @@ pub use self::{
     special::*,
     textual::*,
 };
+use super::EmailFormat;
 use crate::BoxError;
 
 mod content;
@@ -154,6 +155,19 @@ impl Display for Headers {
     }
 }
 
+impl EmailFormat for Headers {
+    fn format<'a>(&'a self, out: &mut impl Extend<Cow<'a, [u8]>>) {
+        for value in &self.headers {
+            out.extend([
+                Cow::Borrowed(value.name.as_bytes()),
+                Cow::Borrowed(b": "),
+                Cow::Borrowed(value.encoded_value.as_bytes()),
+                Cow::Borrowed(b"\r\n"),
+            ]);
+        }
+    }
+}
+
 /// A possible error when converting a `HeaderName` from another type.
 // comes from `http` crate
 #[allow(missing_copy_implementations)]

src/message/mimebody.rs

@@ -1,4 +1,4 @@
-use std::{io::Write, iter::repeat_with};
+use std::{borrow::Cow, iter::repeat_with, sync::Arc};
 
 use mime::Mime;
 
@@ -28,7 +28,7 @@ impl Part {
 }
 
 impl EmailFormat for Part {
-    fn format(&self, out: &mut Vec<u8>) {
+    fn format<'a>(&'a self, out: &mut impl Extend<Cow<'a, [u8]>>) {
         match self {
             Part::Single(part) => part.format(out),
             Part::Multi(part) => part.format(out),
@@ -71,7 +71,7 @@ impl SinglePartBuilder {
 
         SinglePart {
             headers: self.headers,
-            body: body.into_vec(),
+            body: body.into_inner(),
         }
     }
 }
@@ -100,7 +100,7 @@ impl Default for SinglePartBuilder {
 #[derive(Debug, Clone)]
 pub struct SinglePart {
     headers: Headers,
-    body: Vec<u8>,
+    body: Arc<[u8]>,
 }
 
 impl SinglePart {
@@ -138,24 +138,18 @@ impl SinglePart {
 
     /// Get message content formatted for sending
     pub fn formatted(&self) -> Vec<u8> {
-        let mut out = Vec::new();
-        self.format(&mut out);
-        out
-    }
-
-    /// Format only the signlepart body
-    fn format_body(&self, out: &mut Vec<u8>) {
-        out.extend_from_slice(&self.body);
-        out.extend_from_slice(b"\r\n");
+        self.format_to_vec()
     }
 }
 
 impl EmailFormat for SinglePart {
-    fn format(&self, out: &mut Vec<u8>) {
-        write!(out, "{}", self.headers)
-            .expect("A Write implementation panicked while formatting headers");
-        out.extend_from_slice(b"\r\n");
-        self.format_body(out);
+    fn format<'a>(&'a self, out: &mut impl Extend<Cow<'a, [u8]>>) {
+        self.headers.format(out);
+        out.extend([
+            Cow::Borrowed("\r\n".as_bytes()),
+            Cow::Borrowed(&self.body),
+            Cow::Borrowed(b"\r\n"),
+        ]);
     }
 }
 
@@ -384,33 +378,36 @@ impl MultiPart {
 
     /// Get message content formatted for SMTP
     pub fn formatted(&self) -> Vec<u8> {
-        let mut out = Vec::new();
-        self.format(&mut out);
-        out
+        self.format_to_vec()
     }
 
     /// Format only the multipart body
-    fn format_body(&self, out: &mut Vec<u8>) {
+    fn format_body<'a>(&'a self, out: &mut impl Extend<Cow<'a, [u8]>>) {
         let boundary = self.boundary();
 
         for part in &self.parts {
-            out.extend_from_slice(b"--");
-            out.extend_from_slice(boundary.as_bytes());
-            out.extend_from_slice(b"\r\n");
+            out.extend([
+                Cow::Borrowed("--".as_bytes()),
+                // FIXME: this clone shouldn't exist
+                Cow::Owned(boundary.clone().into()),
+                Cow::Borrowed("\r\n".as_bytes()),
+            ]);
             part.format(out);
         }
 
-        out.extend_from_slice(b"--");
-        out.extend_from_slice(boundary.as_bytes());
-        out.extend_from_slice(b"--\r\n");
+        out.extend([
+            Cow::Borrowed("--".as_bytes()),
+            Cow::Owned(boundary.into()),
+            Cow::Borrowed("--\r\n".as_bytes()),
+        ]);
     }
 }
 
 impl EmailFormat for MultiPart {
-    fn format(&self, out: &mut Vec<u8>) {
-        write!(out, "{}", self.headers)
-            .expect("A Write implementation panicked while formatting headers");
-        out.extend_from_slice(b"\r\n");
+    fn format<'a>(&'a self, out: &mut impl Extend<Cow<'a, [u8]>>) {
+        self.headers.format(out);
+        out.extend([Cow::Borrowed("\r\n".as_bytes())]);
+
         self.format_body(out);
     }
 }

src/message/mod.rs

@@ -198,7 +198,7 @@
 //! ```
 //! </details>
 
-use std::{io::Write, iter, time::SystemTime};
+use std::{borrow::Cow, iter, sync::Arc, time::SystemTime};
 
 pub use attachment::Attachment;
 pub use body::{Body, IntoBody, MaybeString};
@@ -226,7 +226,23 @@ const DEFAULT_MESSAGE_ID_DOMAIN: &str = "localhost";
 /// Something that can be formatted as an email message
 trait EmailFormat {
     // Use a writer?
-    fn format(&self, out: &mut Vec<u8>);
+    fn format<'a>(&'a self, out: &mut impl Extend<Cow<'a, [u8]>>);
+
+    fn format_to_vec(&self) -> Vec<u8> {
+        struct Formatter(Vec<u8>);
+
+        impl<'a> Extend<Cow<'a, [u8]>> for Formatter {
+            fn extend<T: IntoIterator<Item = Cow<'a, [u8]>>>(&mut self, iter: T) {
+                for chunk in iter {
+                    self.0.extend_from_slice(&chunk);
+                }
+            }
+        }
+
+        let mut formatted = Formatter(Vec::new());
+        self.format(&mut formatted);
+        formatted.0
+    }
 }
 
 /// A builder for messages
@@ -454,7 +470,7 @@ impl MessageBuilder {
         let body = body.into_body(maybe_encoding);
 
         self.headers.set(body.encoding());
-        self.build(MessageBody::Raw(body.into_vec()))
+        self.build(MessageBody::Raw(body.into_inner()))
     }
 
     /// Create message using mime body ([`MultiPart`][self::MultiPart])
@@ -489,7 +505,7 @@ pub struct Message {
 #[derive(Clone, Debug)]
 enum MessageBody {
     Mime(Part),
-    Raw(Vec<u8>),
+    Raw(Arc<[u8]>),
 }
 
 impl Message {
@@ -515,9 +531,7 @@ impl Message {
 
     /// Get message content formatted for SMTP
     pub fn formatted(&self) -> Vec<u8> {
-        let mut out = Vec::new();
-        self.format(&mut out);
-        out
+        self.format_to_vec()
     }
 
     #[cfg(feature = "dkim")]
@@ -593,15 +607,13 @@ impl Message {
 }
 
 impl EmailFormat for Message {
-    fn format(&self, out: &mut Vec<u8>) {
-        write!(out, "{}", self.headers)
-            .expect("A Write implementation panicked while formatting headers");
+    fn format<'a>(&'a self, out: &mut impl Extend<Cow<'a, [u8]>>) {
+        self.headers.format(out);
 
         match &self.body {
             MessageBody::Mime(p) => p.format(out),
             MessageBody::Raw(r) => {
-                out.extend_from_slice(b"\r\n");
-                out.extend_from_slice(r)
+                out.extend([Cow::Borrowed("\r\n".as_bytes()), Cow::Borrowed(r)]);
             }
         }
     }

src/transport/smtp/client/async_connection.rs

@@ -10,21 +10,34 @@ use super::{AsyncNetworkStream, ClientCodec, TlsParameters};
 use crate::{
     transport::smtp::{
         authentication::{Credentials, Mechanism},
-        client::{ConnectionState, ConnectionWrapper},
         commands::{Auth, Data, Ehlo, Mail, Noop, Quit, Rcpt, Starttls},
-        error::{self, Error},
+        error,
+        error::Error,
         extension::{ClientId, Extension, MailBodyParameter, MailParameter, ServerInfo},
         response::{parse_response, Response},
     },
     Envelope,
 };
 
+macro_rules! try_smtp (
+    ($err: expr, $client: ident) => ({
+        match $err {
+            Ok(val) => val,
+            Err(err) => {
+                $client.abort().await;
+                return Err(From::from(err))
+            },
+        }
+    })
+);
+
 /// Structure that implements the SMTP client
 pub struct AsyncSmtpConnection {
     /// TCP stream between client and server
-    stream: ConnectionWrapper<BufReader<AsyncNetworkStream>>,
-    /// Whether QUIT has been sent
-    sent_quit: bool,
+    /// Value is None before connection
+    stream: BufReader<AsyncNetworkStream>,
+    /// Panic state
+    panic: bool,
     /// Information about the server
     server_info: ServerInfo,
 }
@@ -112,8 +125,8 @@ impl AsyncSmtpConnection {
     ) -> Result<AsyncSmtpConnection, Error> {
         let stream = BufReader::new(stream);
         let mut conn = AsyncSmtpConnection {
-            stream: ConnectionWrapper::new(stream),
-            sent_quit: false,
+            stream,
+            panic: false,
             server_info: ServerInfo::default(),
         };
         // TODO log
@@ -157,28 +170,30 @@ impl AsyncSmtpConnection {
             mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime));
         }
 
-        self.command(Mail::new(envelope.from().cloned(), mail_options))
-            .await?;
+        try_smtp!(
+            self.command(Mail::new(envelope.from().cloned(), mail_options))
+                .await,
+            self
+        );
 
         // Recipient
         for to_address in envelope.to() {
-            self.command(Rcpt::new(to_address.clone(), vec![])).await?;
+            try_smtp!(
+                self.command(Rcpt::new(to_address.clone(), vec![])).await,
+                self
+            );
         }
 
         // Data
-        self.command(Data).await?;
+        try_smtp!(self.command(Data).await, self);
 
         // Message content
-        let result = self.message(email).await?;
+        let result = try_smtp!(self.message(email).await, self);
         Ok(result)
     }
 
     pub fn has_broken(&self) -> bool {
-        self.sent_quit
-            || matches!(
-                self.stream.state(),
-                ConnectionState::BrokenConnection | ConnectionState::BrokenResponse
-            )
+        self.panic
     }
 
     pub fn can_starttls(&self) -> bool {
@@ -198,14 +213,12 @@ impl AsyncSmtpConnection {
         hello_name: &ClientId,
     ) -> Result<(), Error> {
         if self.server_info.supports_feature(Extension::StartTls) {
-            self.command(Starttls).await?;
-            self.stream
-                .async_op(|stream| stream.get_mut().upgrade_tls(tls_parameters))
-                .await?;
+            try_smtp!(self.command(Starttls).await, self);
+            self.stream.get_mut().upgrade_tls(tls_parameters).await?;
             #[cfg(feature = "tracing")]
             tracing::debug!("connection encrypted");
             // Send EHLO again
-            self.ehlo(hello_name).await?;
+            try_smtp!(self.ehlo(hello_name).await, self);
             Ok(())
         } else {
             Err(error::client("STARTTLS is not supported on this server"))
@@ -214,39 +227,32 @@ impl AsyncSmtpConnection {
 
     /// Send EHLO and update server info
     async fn ehlo(&mut self, hello_name: &ClientId) -> Result<(), Error> {
-        let ehlo_response = self.command(Ehlo::new(hello_name.clone())).await?;
-        self.server_info = ServerInfo::from_response(&ehlo_response)?;
+        let ehlo_response = try_smtp!(self.command(Ehlo::new(hello_name.clone())).await, self);
+        self.server_info = try_smtp!(ServerInfo::from_response(&ehlo_response), self);
         Ok(())
     }
 
     pub async fn quit(&mut self) -> Result<Response, Error> {
-        self.sent_quit = true;
-        self.command(Quit).await
+        Ok(try_smtp!(self.command(Quit).await, self))
     }
 
     pub async fn abort(&mut self) {
         // Only try to quit if we are not already broken
-        // `write` already rejects writes if the connection state if bad
-        if !self.sent_quit {
-            let _ = self.quit().await;
-        }
-
-        if !matches!(self.stream.state(), ConnectionState::BrokenConnection) {
-            let _ = self
-                .stream
-                .async_op(|stream| async { stream.close().await.map_err(error::network) })
-                .await;
+        if !self.panic {
+            self.panic = true;
+            let _ = self.command(Quit).await;
         }
+        let _ = self.stream.close().await;
     }
 
     /// Sets the underlying stream
     pub fn set_stream(&mut self, stream: AsyncNetworkStream) {
-        self.stream = ConnectionWrapper::new(BufReader::new(stream));
+        self.stream = BufReader::new(stream);
     }
 
     /// Tells if the underlying stream is currently encrypted
     pub fn is_encrypted(&self) -> bool {
-        self.stream.get_ref().get_ref().is_encrypted()
+        self.stream.get_ref().is_encrypted()
     }
 
     /// Checks if the server is connected using the NOOP SMTP command
@@ -273,13 +279,15 @@ impl AsyncSmtpConnection {
 
         while challenges > 0 && response.has_code(334) {
             challenges -= 1;
-            response = self
-                .command(Auth::new_from_response(
+            response = try_smtp!(
+                self.command(Auth::new_from_response(
                     mechanism,
                     credentials.clone(),
                     &response,
                 )?)
-                .await?;
+                .await,
+                self
+            );
         }
 
         if challenges == 0 {
@@ -308,17 +316,15 @@ impl AsyncSmtpConnection {
     /// Writes a string to the server
     async fn write(&mut self, string: &[u8]) -> Result<(), Error> {
         self.stream
-            .async_op(|stream| async {
-                stream
-                    .get_mut()
-                    .write_all(string)
-                    .await
-                    .map_err(error::network)
-            })
-            .await?;
+            .get_mut()
+            .write_all(string)
+            .await
+            .map_err(error::network)?;
         self.stream
-            .async_op(|stream| async { stream.get_mut().flush().await.map_err(error::network) })
-            .await?;
+            .get_mut()
+            .flush()
+            .await
+            .map_err(error::network)?;
 
         #[cfg(feature = "tracing")]
         tracing::debug!("Wrote: {}", escape_crlf(&String::from_utf8_lossy(string)));
@@ -331,10 +337,9 @@ impl AsyncSmtpConnection {
 
         while self
             .stream
-            .async_op(|stream| async {
-                stream.read_line(&mut buffer).await.map_err(error::network)
-            })
-            .await?
+            .read_line(&mut buffer)
+            .await
+            .map_err(error::network)?
             > 0
         {
             #[cfg(feature = "tracing")]
@@ -351,12 +356,10 @@ impl AsyncSmtpConnection {
                     }
                 }
                 Err(nom::Err::Failure(e)) => {
-                    self.stream.set_state(ConnectionState::BrokenResponse);
                     return Err(error::response(e.to_string()));
                 }
                 Err(nom::Err::Incomplete(_)) => { /* read more */ }
                 Err(nom::Err::Error(e)) => {
-                    self.stream.set_state(ConnectionState::BrokenResponse);
                     return Err(error::response(e.to_string()));
                 }
             }
@@ -368,12 +371,12 @@ impl AsyncSmtpConnection {
     /// The X509 certificate of the server (DER encoded)
     #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
     pub fn peer_certificate(&self) -> Result<Vec<u8>, Error> {
-        self.stream.get_ref().get_ref().peer_certificate()
+        self.stream.get_ref().peer_certificate()
     }
 
     /// All the X509 certificates of the chain (DER encoded)
     #[cfg(any(feature = "rustls-tls", feature = "boring-tls"))]
     pub fn certificate_chain(&self) -> Result<Vec<Vec<u8>>, Error> {
-        self.stream.get_ref().get_ref().certificate_chain()
+        self.stream.get_ref().certificate_chain()
     }
 }

src/transport/smtp/client/connection.rs

@@ -7,25 +7,38 @@ use std::{
 
 #[cfg(feature = "tracing")]
 use super::escape_crlf;
-use super::{ClientCodec, ConnectionWrapper, NetworkStream, TlsParameters};
+use super::{ClientCodec, NetworkStream, TlsParameters};
 use crate::{
     address::Envelope,
     transport::smtp::{
         authentication::{Credentials, Mechanism},
-        client::ConnectionState,
         commands::{Auth, Data, Ehlo, Mail, Noop, Quit, Rcpt, Starttls},
-        error::{self, Error},
+        error,
+        error::Error,
         extension::{ClientId, Extension, MailBodyParameter, MailParameter, ServerInfo},
         response::{parse_response, Response},
     },
 };
 
+macro_rules! try_smtp (
+    ($err: expr, $client: ident) => ({
+        match $err {
+            Ok(val) => val,
+            Err(err) => {
+                $client.abort();
+                return Err(From::from(err))
+            },
+        }
+    })
+);
+
 /// Structure that implements the SMTP client
 pub struct SmtpConnection {
     /// TCP stream between client and server
-    stream: ConnectionWrapper<BufReader<NetworkStream>>,
-    /// Whether QUIT has been sent
-    sent_quit: bool,
+    /// Value is None before connection
+    stream: BufReader<NetworkStream>,
+    /// Panic state
+    panic: bool,
     /// Information about the server
     server_info: ServerInfo,
 }
@@ -51,8 +64,8 @@ impl SmtpConnection {
         let stream = NetworkStream::connect(server, timeout, tls_parameters, local_address)?;
         let stream = BufReader::new(stream);
         let mut conn = SmtpConnection {
-            stream: ConnectionWrapper::new(stream),
-            sent_quit: false,
+            stream,
+            panic: false,
             server_info: ServerInfo::default(),
         };
         conn.set_timeout(timeout).map_err(error::network)?;
@@ -97,27 +110,26 @@ impl SmtpConnection {
             mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime));
         }
 
-        self.command(Mail::new(envelope.from().cloned(), mail_options))?;
+        try_smtp!(
+            self.command(Mail::new(envelope.from().cloned(), mail_options)),
+            self
+        );
 
         // Recipient
         for to_address in envelope.to() {
-            self.command(Rcpt::new(to_address.clone(), vec![]))?;
+            try_smtp!(self.command(Rcpt::new(to_address.clone(), vec![])), self);
         }
 
         // Data
-        self.command(Data)?;
+        try_smtp!(self.command(Data), self);
 
         // Message content
-        let result = self.message(email)?;
+        let result = try_smtp!(self.message(email), self);
         Ok(result)
     }
 
     pub fn has_broken(&self) -> bool {
-        self.sent_quit
-            || matches!(
-                self.stream.state(),
-                ConnectionState::BrokenConnection | ConnectionState::BrokenResponse
-            )
+        self.panic
     }
 
     pub fn can_starttls(&self) -> bool {
@@ -133,13 +145,12 @@ impl SmtpConnection {
         if self.server_info.supports_feature(Extension::StartTls) {
             #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
             {
-                self.command(Starttls)?;
-                self.stream
-                    .sync_op(|stream| stream.get_mut().upgrade_tls(tls_parameters))?;
+                try_smtp!(self.command(Starttls), self);
+                self.stream.get_mut().upgrade_tls(tls_parameters)?;
                 #[cfg(feature = "tracing")]
                 tracing::debug!("connection encrypted");
                 // Send EHLO again
-                self.ehlo(hello_name)?;
+                try_smtp!(self.ehlo(hello_name), self);
                 Ok(())
             }
             #[cfg(not(any(
@@ -157,47 +168,38 @@ impl SmtpConnection {
 
     /// Send EHLO and update server info
     fn ehlo(&mut self, hello_name: &ClientId) -> Result<(), Error> {
-        let ehlo_response = self.command(Ehlo::new(hello_name.clone()))?;
-        self.server_info = ServerInfo::from_response(&ehlo_response)?;
+        let ehlo_response = try_smtp!(self.command(Ehlo::new(hello_name.clone())), self);
+        self.server_info = try_smtp!(ServerInfo::from_response(&ehlo_response), self);
         Ok(())
     }
 
     pub fn quit(&mut self) -> Result<Response, Error> {
-        self.sent_quit = true;
-        self.command(Quit)
+        Ok(try_smtp!(self.command(Quit), self))
     }
 
     pub fn abort(&mut self) {
         // Only try to quit if we are not already broken
-        // `write` already rejects writes if the connection state if bad
-        if !self.sent_quit {
-            let _ = self.quit();
-        }
-
-        if !matches!(self.stream.state(), ConnectionState::BrokenConnection) {
-            let _ = self.stream.sync_op(|stream| {
-                stream
-                    .get_mut()
-                    .shutdown(std::net::Shutdown::Both)
-                    .map_err(error::network)
-            });
+        if !self.panic {
+            self.panic = true;
+            let _ = self.command(Quit);
         }
+        let _ = self.stream.get_mut().shutdown(std::net::Shutdown::Both);
     }
 
     /// Sets the underlying stream
     pub fn set_stream(&mut self, stream: NetworkStream) {
-        self.stream = ConnectionWrapper::new(BufReader::new(stream));
+        self.stream = BufReader::new(stream);
     }
 
     /// Tells if the underlying stream is currently encrypted
     pub fn is_encrypted(&self) -> bool {
-        self.stream.get_ref().get_ref().is_encrypted()
+        self.stream.get_ref().is_encrypted()
     }
 
     /// Set timeout
     pub fn set_timeout(&mut self, duration: Option<Duration>) -> io::Result<()> {
-        self.stream.get_mut().get_mut().set_read_timeout(duration)?;
-        self.stream.get_mut().get_mut().set_write_timeout(duration)
+        self.stream.get_mut().set_read_timeout(duration)?;
+        self.stream.get_mut().set_write_timeout(duration)
     }
 
     /// Checks if the server is connected using the NOOP SMTP command
@@ -222,11 +224,14 @@ impl SmtpConnection {
 
         while challenges > 0 && response.has_code(334) {
             challenges -= 1;
-            response = self.command(Auth::new_from_response(
-                mechanism,
-                credentials.clone(),
-                &response,
-            )?)?;
+            response = try_smtp!(
+                self.command(Auth::new_from_response(
+                    mechanism,
+                    credentials.clone(),
+                    &response,
+                )?),
+                self
+            );
         }
 
         if challenges == 0 {
@@ -256,9 +261,10 @@ impl SmtpConnection {
     /// Writes a string to the server
     fn write(&mut self, string: &[u8]) -> Result<(), Error> {
         self.stream
-            .sync_op(|stream| stream.get_mut().write_all(string).map_err(error::network))?;
-        self.stream
-            .sync_op(|stream| stream.get_mut().flush().map_err(error::network))?;
+            .get_mut()
+            .write_all(string)
+            .map_err(error::network)?;
+        self.stream.get_mut().flush().map_err(error::network)?;
 
         #[cfg(feature = "tracing")]
         tracing::debug!("Wrote: {}", escape_crlf(&String::from_utf8_lossy(string)));
@@ -269,11 +275,7 @@ impl SmtpConnection {
     pub fn read_response(&mut self) -> Result<Response, Error> {
         let mut buffer = String::with_capacity(100);
 
-        while self
-            .stream
-            .sync_op(|stream| stream.read_line(&mut buffer).map_err(error::network))?
-            > 0
-        {
+        while self.stream.read_line(&mut buffer).map_err(error::network)? > 0 {
             #[cfg(feature = "tracing")]
             tracing::debug!("<< {}", escape_crlf(&buffer));
             match parse_response(&buffer) {
@@ -288,12 +290,10 @@ impl SmtpConnection {
                     };
                 }
                 Err(nom::Err::Failure(e)) => {
-                    self.stream.set_state(ConnectionState::BrokenResponse);
                     return Err(error::response(e.to_string()));
                 }
                 Err(nom::Err::Incomplete(_)) => { /* read more */ }
                 Err(nom::Err::Error(e)) => {
-                    self.stream.set_state(ConnectionState::BrokenResponse);
                     return Err(error::response(e.to_string()));
                 }
             }
@@ -305,12 +305,12 @@ impl SmtpConnection {
     /// The X509 certificate of the server (DER encoded)
     #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
     pub fn peer_certificate(&self) -> Result<Vec<u8>, Error> {
-        self.stream.get_ref().get_ref().peer_certificate()
+        self.stream.get_ref().peer_certificate()
     }
 
     /// All the X509 certificates of the chain (DER encoded)
     #[cfg(any(feature = "rustls-tls", feature = "boring-tls"))]
     pub fn certificate_chain(&self) -> Result<Vec<Vec<u8>>, Error> {
-        self.stream.get_ref().get_ref().certificate_chain()
+        self.stream.get_ref().certificate_chain()
     }
 }

src/transport/smtp/client/mod.rs

@@ -24,8 +24,6 @@
 
 #[cfg(feature = "serde")]
 use std::fmt::Debug;
-#[cfg(any(feature = "tokio1", feature = "async-std1"))]
-use std::future::Future;
 
 #[cfg(any(feature = "tokio1", feature = "async-std1"))]
 pub use self::async_connection::AsyncSmtpConnection;
@@ -42,7 +40,6 @@ pub use self::{
     connection::SmtpConnection,
     tls::{Certificate, CertificateStore, Identity, Tls, TlsParameters, TlsParametersBuilder},
 };
-use super::{error, Error};
 
 #[cfg(any(feature = "tokio1", feature = "async-std1"))]
 mod async_connection;
@@ -52,99 +49,6 @@ mod connection;
 mod net;
 mod tls;
 
-#[derive(Debug)]
-pub(super) struct ConnectionWrapper<C> {
-    conn: C,
-    state: ConnectionState,
-}
-
-impl<C> ConnectionWrapper<C> {
-    pub(super) fn new(conn: C) -> Self {
-        Self {
-            conn,
-            state: ConnectionState::ProbablyConnected,
-        }
-    }
-
-    pub(super) fn get_ref(&self) -> &C {
-        &self.conn
-    }
-
-    pub(super) fn get_mut(&mut self) -> &mut C {
-        &mut self.conn
-    }
-
-    pub(super) fn state(&self) -> ConnectionState {
-        self.state
-    }
-
-    pub(super) fn set_state(&mut self, state: ConnectionState) {
-        self.state = state;
-    }
-
-    pub(super) fn sync_op<F, T>(&mut self, f: F) -> Result<T, Error>
-    where
-        F: FnOnce(&mut C) -> Result<T, Error>,
-    {
-        if !matches!(
-            self.state,
-            ConnectionState::ProbablyConnected | ConnectionState::BrokenResponse
-        ) {
-            return Err(error::client(
-                "attempted to send operation to broken connection",
-            ));
-        }
-
-        self.state = ConnectionState::Writing;
-        match f(&mut self.conn) {
-            Ok(t) => {
-                self.state = ConnectionState::ProbablyConnected;
-                Ok(t)
-            }
-            Err(err) => {
-                self.state = ConnectionState::BrokenConnection;
-                Err(err)
-            }
-        }
-    }
-
-    #[cfg(any(feature = "tokio1", feature = "async-std1"))]
-    pub(super) async fn async_op<'a, F, Fut, T>(&'a mut self, f: F) -> Result<T, Error>
-    where
-        F: FnOnce(&'a mut C) -> Fut,
-        Fut: Future<Output = Result<T, Error>>,
-    {
-        if !matches!(
-            self.state,
-            ConnectionState::ProbablyConnected | ConnectionState::BrokenResponse
-        ) {
-            return Err(error::client(
-                "attempted to send operation to broken connection",
-            ));
-        }
-
-        self.state = ConnectionState::Writing;
-        match f(&mut self.conn).await {
-            Ok(t) => {
-                self.state = ConnectionState::ProbablyConnected;
-                Ok(t)
-            }
-            Err(err) => {
-                self.state = ConnectionState::BrokenConnection;
-                Err(err)
-            }
-        }
-    }
-}
-
-#[derive(Debug, Copy, Clone)]
-pub(super) enum ConnectionState {
-    ProbablyConnected,
-    Writing,
-    BrokenResponse,
-    BrokenConnection,
-}
-
 /// The codec used for transparency
 #[derive(Debug)]
 struct ClientCodec {