refactor!: use try_next() more

I have looked through the code
to make sure errors are not ignored,
especially in loops.
Ignoring errors in loops
when reading from streams
may result in infinite loop
reading the same error over and over again.
No bugs found,
but I refactored reading from streams
to use `try_next()` more
to bubble up the errors with `?`
as soon as possible.

This is a breaking change since
`read_response()` is resultified to return `Result<Option<_>>`
instead of `Option<Result<_>>`.
`read_response()` is a public interface
that is used by library users
to read the banner.

Author: link2xt

examples/src/bin/integration.rs

@@ -31,7 +31,7 @@ async fn session(user: &str) -> Result<Session<async_native_tls::TlsStream<TcpSt
     let mut client = async_imap::Client::new(tls_stream);
     let _greeting = client
         .read_response()
-        .await
+        .await?
         .context("unexpected end of stream, expected greeting")?;
 
     let session = client
@@ -48,7 +48,7 @@ async fn _connect_insecure_then_secure() -> Result<()> {
     let mut client = async_imap::Client::new(tcp_stream);
     let _greeting = client
         .read_response()
-        .await
+        .await?
         .context("unexpected end of stream, expected greeting")?;
     client.run_command_and_check_ok("STARTTLS", None).await?;
     let stream = client.into_inner();

src/client.rs

@@ -10,7 +10,7 @@ use async_std::io::{Read, Write, WriteExt};
 use base64::Engine as _;
 use extensions::id::{format_identification, parse_id};
 use extensions::quota::parse_get_quota_root;
-use futures::{io, Stream, StreamExt};
+use futures::{io, Stream, TryStreamExt};
 use imap_proto::{Metadata, RequestId, Response};
 #[cfg(feature = "runtime-tokio")]
 use tokio::io::{AsyncRead as Read, AsyncWrite as Write, AsyncWriteExt};
@@ -122,7 +122,7 @@ macro_rules! ok_or_unauth_client_err {
     ($r:expr, $self:expr) => {
         match $r {
             Ok(o) => o,
-            Err(e) => return Err((e, $self)),
+            Err(e) => return Err((e.into(), $self)),
         }
     };
 }
@@ -262,42 +262,37 @@ impl<T: Read + Write + Unpin + fmt::Debug + Send> Client<T> {
         // explicit match blocks neccessary to convert error to tuple and not bind self too
         // early (see also comment on `login`)
         loop {
-            if let Some(res) = self.read_response().await {
-                let res = ok_or_unauth_client_err!(res.map_err(Into::into), self);
-                match res.parsed() {
-                    Response::Continue { information, .. } => {
-                        let challenge = if let Some(text) = information {
-                            ok_or_unauth_client_err!(
-                                base64::engine::general_purpose::STANDARD
-                                    .decode(text.as_ref())
-                                    .map_err(|e| Error::Parse(ParseError::Authentication(
-                                        (*text).to_string(),
-                                        Some(e)
-                                    ))),
-                                self
-                            )
-                        } else {
-                            Vec::new()
-                        };
-                        let raw_response = &mut authenticator.process(&challenge);
-                        let auth_response =
-                            base64::engine::general_purpose::STANDARD.encode(raw_response);
-
-                        ok_or_unauth_client_err!(
-                            self.conn.run_command_untagged(&auth_response).await,
-                            self
-                        );
-                    }
-                    _ => {
+            let Some(res) = ok_or_unauth_client_err!(self.read_response().await, self) else {
+                return Err((Error::ConnectionLost, self));
+            };
+            match res.parsed() {
+                Response::Continue { information, .. } => {
+                    let challenge = if let Some(text) = information {
                         ok_or_unauth_client_err!(
-                            self.check_done_ok_from(&id, None, res).await,
+                            base64::engine::general_purpose::STANDARD
+                                .decode(text.as_ref())
+                                .map_err(|e| Error::Parse(ParseError::Authentication(
+                                    (*text).to_string(),
+                                    Some(e)
+                                ))),
                             self
-                        );
-                        return Ok(Session::new(self.conn));
-                    }
+                        )
+                    } else {
+                        Vec::new()
+                    };
+                    let raw_response = &mut authenticator.process(&challenge);
+                    let auth_response =
+                        base64::engine::general_purpose::STANDARD.encode(raw_response);
+
+                    ok_or_unauth_client_err!(
+                        self.conn.run_command_untagged(&auth_response).await,
+                        self
+                    );
+                }
+                _ => {
+                    ok_or_unauth_client_err!(self.check_done_ok_from(&id, None, res).await, self);
+                    return Ok(Session::new(self.conn));
                 }
-            } else {
-                return Err((Error::ConnectionLost, self));
             }
         }
     }
@@ -975,12 +970,13 @@ impl<T: Read + Write + Unpin + fmt::Debug + Send> Session<T> {
                 mailbox_pattern.unwrap_or("\"\"")
             ))
             .await?;
-
-        Ok(parse_names(
+        let names = parse_names(
             &mut self.conn.stream,
             self.unsolicited_responses_tx.clone(),
             id,
-        ))
+        );
+
+        Ok(names)
     }
 
     /// The [`LSUB` command](https://tools.ietf.org/html/rfc3501#section-6.3.9) returns a subset of
@@ -1136,23 +1132,20 @@ impl<T: Read + Write + Unpin + fmt::Debug + Send> Session<T> {
             ))
             .await?;
 
-        match self.read_response().await {
-            Some(Ok(res)) => {
-                if let Response::Continue { .. } = res.parsed() {
-                    self.stream.as_mut().write_all(content).await?;
-                    self.stream.as_mut().write_all(b"\r\n").await?;
-                    self.stream.flush().await?;
-                    self.conn
-                        .check_done_ok(&id, Some(self.unsolicited_responses_tx.clone()))
-                        .await?;
-                    Ok(())
-                } else {
-                    Err(Error::Append)
-                }
-            }
-            Some(Err(err)) => Err(err.into()),
-            _ => Err(Error::Append),
-        }
+        let Some(res) = self.read_response().await? else {
+            return Err(Error::Append);
+        };
+        let Response::Continue { .. } = res.parsed() else {
+            return Err(Error::Append);
+        };
+
+        self.stream.as_mut().write_all(content).await?;
+        self.stream.as_mut().write_all(b"\r\n").await?;
+        self.stream.flush().await?;
+        self.conn
+            .check_done_ok(&id, Some(self.unsolicited_responses_tx.clone()))
+            .await?;
+        Ok(())
     }
 
     /// The [`SEARCH` command](https://tools.ietf.org/html/rfc3501#section-6.4.4) searches the
@@ -1352,7 +1345,7 @@ impl<T: Read + Write + Unpin + fmt::Debug + Send> Session<T> {
     }
 
     /// Read the next response on the connection.
-    pub async fn read_response(&mut self) -> Option<io::Result<ResponseData>> {
+    pub async fn read_response(&mut self) -> io::Result<Option<ResponseData>> {
         self.conn.read_response().await
     }
 }
@@ -1377,8 +1370,8 @@ impl<T: Read + Write + Unpin + fmt::Debug> Connection<T> {
     }
 
     /// Read the next response on the connection.
-    pub async fn read_response(&mut self) -> Option<io::Result<ResponseData>> {
-        self.stream.next().await
+    pub async fn read_response(&mut self) -> io::Result<Option<ResponseData>> {
+        self.stream.try_next().await
     }
 
     pub(crate) async fn run_command_untagged(&mut self, command: &str) -> Result<()> {
@@ -1415,8 +1408,8 @@ impl<T: Read + Write + Unpin + fmt::Debug> Connection<T> {
         id: &RequestId,
         unsolicited: Option<channel::Sender<UnsolicitedResponse>>,
     ) -> Result<()> {
-        if let Some(first_res) = self.stream.next().await {
-            self.check_done_ok_from(id, unsolicited, first_res?).await
+        if let Some(first_res) = self.stream.try_next().await? {
+            self.check_done_ok_from(id, unsolicited, first_res).await
         } else {
             Err(Error::ConnectionLost)
         }
@@ -1447,11 +1440,10 @@ impl<T: Read + Write + Unpin + fmt::Debug> Connection<T> {
                 handle_unilateral(response, unsolicited);
             }
 
-            if let Some(res) = self.stream.next().await {
-                response = res?;
-            } else {
+            let Some(res) = self.stream.try_next().await? else {
                 return Err(Error::ConnectionLost);
-            }
+            };
+            response = res;
         }
     }
 
@@ -1495,6 +1487,7 @@ mod tests {
     use std::future::Future;
 
     use async_std::sync::{Arc, Mutex};
+    use futures::StreamExt;
     use imap_proto::Status;
 
     macro_rules! mock_client {
@@ -1555,7 +1548,7 @@ mod tests {
     async fn readline_eof() {
         let mock_stream = MockStream::default().with_eof();
         let mut client = mock_client!(mock_stream);
-        let res = client.read_response().await;
+        let res = client.read_response().await.unwrap();
         assert!(res.is_none());
     }
 
@@ -2117,7 +2110,7 @@ mod tests {
                 .unwrap();
 
             // Unexpected EOF.
-            let err = fetch_result.next().await.unwrap().unwrap_err();
+            let err = fetch_result.try_next().await.unwrap_err();
             let Error::Io(io_err) = err else {
                 panic!("Unexpected error type: {err}")
             };

src/extensions/id.rs

@@ -43,10 +43,9 @@ pub(crate) async fn parse_id<T: Stream<Item = io::Result<ResponseData>> + Unpin>
     let mut id = None;
     while let Some(resp) = stream
         .take_while(|res| filter(res, &command_tag))
-        .next()
-        .await
+        .try_next()
+        .await?
     {
-        let resp = resp?;
         match resp.parsed() {
             Response::Id(res) => {
                 id = res.as_ref().map(|m| {

src/extensions/idle.rs

@@ -182,8 +182,7 @@ impl<T: Read + Write + Unpin + fmt::Debug + Send> Handle<T> {
     pub async fn init(&mut self) -> Result<()> {
         let id = self.session.run_command("IDLE").await?;
         self.id = Some(id);
-        while let Some(res) = self.session.stream.next().await {
-            let res = res?;
+        while let Some(res) = self.session.stream.try_next().await? {
             match res.parsed() {
                 Response::Continue { .. } => {
                     return Ok(());

src/extensions/quota.rs

@@ -23,10 +23,9 @@ pub(crate) async fn parse_get_quota<T: Stream<Item = io::Result<ResponseData>> +
     let mut quota = None;
     while let Some(resp) = stream
         .take_while(|res| filter(res, &command_tag))
-        .next()
-        .await
+        .try_next()
+        .await?
     {
-        let resp = resp?;
         match resp.parsed() {
             Response::Quota(q) => quota = Some(q.clone().into()),
             _ => {
@@ -53,10 +52,9 @@ pub(crate) async fn parse_get_quota_root<T: Stream<Item = io::Result<ResponseDat
 
     while let Some(resp) = stream
         .take_while(|res| filter(res, &command_tag))
-        .next()
-        .await
+        .try_next()
+        .await?
     {
-        let resp = resp?;
         match resp.parsed() {
             Response::QuotaRoot(qr) => {
                 roots.push(qr.clone().into());

src/parse.rs

@@ -99,8 +99,7 @@ pub(crate) async fn parse_status<T: Stream<Item = io::Result<ResponseData>> + Un
 ) -> Result<Mailbox> {
     let mut mbox = Mailbox::default();
 
-    while let Some(resp) = stream.next().await {
-        let resp = resp?;
+    while let Some(resp) = stream.try_next().await? {
         match resp.parsed() {
             Response::Done {
                 tag,
@@ -192,10 +191,9 @@ pub(crate) async fn parse_capabilities<T: Stream<Item = io::Result<ResponseData>
 
     while let Some(resp) = stream
         .take_while(|res| filter(res, &command_tag))
-        .next()
-        .await
+        .try_next()
+        .await?
     {
-        let resp = resp?;
         match resp.parsed() {
             Response::Capabilities(cs) => {
                 for c in cs {
@@ -218,10 +216,9 @@ pub(crate) async fn parse_noop<T: Stream<Item = io::Result<ResponseData>> + Unpi
 ) -> Result<()> {
     while let Some(resp) = stream
         .take_while(|res| filter(res, &command_tag))
-        .next()
-        .await
+        .try_next()
+        .await?
     {
-        let resp = resp?;
         handle_unilateral(resp, unsolicited.clone());
     }
 
@@ -235,8 +232,7 @@ pub(crate) async fn parse_mailbox<T: Stream<Item = io::Result<ResponseData>> + U
 ) -> Result<Mailbox> {
     let mut mailbox = Mailbox::default();
 
-    while let Some(resp) = stream.next().await {
-        let resp = resp?;
+    while let Some(resp) = stream.try_next().await? {
         match resp.parsed() {
             Response::Done {
                 tag,
@@ -345,10 +341,9 @@ pub(crate) async fn parse_ids<T: Stream<Item = io::Result<ResponseData>> + Unpin
 
     while let Some(resp) = stream
         .take_while(|res| filter(res, &command_tag))
-        .next()
-        .await
+        .try_next()
+        .await?
     {
-        let resp = resp?;
         match resp.parsed() {
             Response::MailboxData(MailboxDatum::Search(cs)) => {
                 for c in cs {
@@ -374,10 +369,9 @@ pub(crate) async fn parse_metadata<T: Stream<Item = io::Result<ResponseData>> +
     let mut res_values = Vec::new();
     while let Some(resp) = stream
         .take_while(|res| filter(res, &command_tag))
-        .next()
-        .await
+        .try_next()
+        .await?
     {
-        let resp = resp?;
         match resp.parsed() {
             // METADATA Response with Values
             // <https://datatracker.ietf.org/doc/html/rfc5464.html#section-4.4.1>