split RuntimeShutdownError into its own error kind

Author: Keruspe

src/connection.rs

@@ -21,7 +21,7 @@ use amq_protocol::frame::{AMQPFrame, ProtocolVersion};
 use async_rs::{Runtime, traits::*};
 use async_trait::async_trait;
 use futures_core::Stream;
-use std::{fmt, io, sync::Arc};
+use std::{fmt, sync::Arc};
 use tracing::trace;
 
 /// A TCP connection to the AMQP server.
@@ -315,10 +315,9 @@ impl Connect for AMQPUri {
             self,
             runtime,
             async move |uri, runtime| {
-                Ok(
-                    AMQPUriTcpExt::connect_with_config_async(&uri, config.as_ref(), &runtime)
-                        .await?,
-                )
+                AMQPUriTcpExt::connect_with_config_async(&uri, config.as_ref(), &runtime)
+                    .await
+                    .map_err(Error::io)
             },
             options,
         )

src/connection_builder.rs

@@ -4,7 +4,6 @@ use crate::{
 };
 
 use async_rs::{Runtime, traits::*};
-use std::io;
 
 #[derive(Debug)]
 pub struct ConnectionBuilder<RK: RuntimeKit + Send + Sync + Clone + 'static> {

src/error.rs

@@ -33,6 +33,7 @@ pub enum ErrorKind {
     InvalidConnectionState(ConnectionState),
 
     IOError(Arc<io::Error>),
+    RuntimeShutdownError(Arc<io::Error>),
     ParsingError(ParserError),
     ProtocolError(AMQPError),
     SerialisationError(Arc<GenError>),
@@ -43,7 +44,15 @@ pub enum ErrorKind {
 
 impl Error {
     pub(crate) fn other<E: Into<Box<dyn error::Error + Send + Sync>>>(error: E) -> Self {
-        ErrorKind::IOError(Arc::new(io::Error::other(error))).into()
+        io::Error::other(error).into()
+    }
+
+    pub(crate) fn io(error: io::Error) -> Self {
+        if io_error_is_runtime_shutdown(&error) {
+            ErrorKind::RuntimeShutdownError(Arc::new(error)).into()
+        } else {
+            error.into()
+        }
     }
 
     pub fn kind(&self) -> &ErrorKind {
@@ -79,6 +88,13 @@ impl Error {
         if let ErrorKind::IOError(_) = self.kind() {
             return true;
         }
+        self.is_runtime_shutdown_error()
+    }
+
+    pub fn is_runtime_shutdown_error(&self) -> bool {
+        if let ErrorKind::RuntimeShutdownError(_) = self.kind() {
+            return true;
+        }
         false
     }
 
@@ -107,13 +123,6 @@ impl Error {
         false
     }
 
-    pub fn is_runtime_shutdown_error(&self) -> bool {
-        if let ErrorKind::IOError(e) = self.kind() {
-            return io_error_is_runtime_shutdown(e);
-        }
-        false
-    }
-
     pub fn can_be_recovered(&self) -> bool {
         match self.kind() {
             ErrorKind::ChannelsLimitReached => false,
@@ -123,7 +132,8 @@ impl Error {
             ErrorKind::InvalidChannelState(..) => true,
             ErrorKind::InvalidConnectionState(_) => true,
 
-            ErrorKind::IOError(e) => !io_error_is_runtime_shutdown(e),
+            ErrorKind::IOError(_) => true,
+            ErrorKind::RuntimeShutdownError(_) => false,
             ErrorKind::ParsingError(_) => false,
             ErrorKind::ProtocolError(_) => true,
             ErrorKind::SerialisationError(_) => false,
@@ -166,6 +176,7 @@ impl fmt::Display for Error {
             }
 
             ErrorKind::IOError(e) => write!(f, "IO error: {e}"),
+            ErrorKind::RuntimeShutdownError(e) => write!(f, "runtime shutdown error: {e}"),
             ErrorKind::ParsingError(e) => write!(f, "failed to parse: {e}"),
             ErrorKind::ProtocolError(e) => write!(f, "protocol error: {e}"),
             ErrorKind::SerialisationError(e) => write!(f, "failed to serialise: {e}"),
@@ -182,6 +193,7 @@ impl error::Error for Error {
     fn source(&self) -> Option<&(dyn error::Error + 'static)> {
         match self.kind() {
             ErrorKind::IOError(e) => Some(&**e),
+            ErrorKind::RuntimeShutdownError(e) => Some(&**e),
             ErrorKind::ParsingError(e) => Some(e),
             ErrorKind::ProtocolError(e) => Some(e),
             ErrorKind::SerialisationError(e) => Some(&**e),
@@ -229,6 +241,10 @@ impl PartialEq for Error {
                 error!("Unable to compare lapin::ErrorKind::IOError");
                 false
             }
+            (RuntimeShutdownError(_), RuntimeShutdownError(_)) => {
+                error!("Unable to compare lapin::ErrorKind::RuntimeShutdownError");
+                false
+            }
             (ParsingError(left_inner), ParsingError(right_inner)) => left_inner == right_inner,
             (ProtocolError(left_inner), ProtocolError(right_inner)) => left_inner == right_inner,
             (SerialisationError(_), SerialisationError(_)) => {

src/io_loop.rs

@@ -270,7 +270,7 @@ impl<
                         error!(?err, "Failed to close IO stream");
                     }
                 res
-            })?;
+            }).map_err(Error::io)?;
         waker.wake();
         Ok(handle)
     }
@@ -391,7 +391,7 @@ impl<
         stream: Pin<&mut T>,
         writable_context: &mut Context<'_>,
     ) -> Result<()> {
-        let res = stream.poll_flush(writable_context)?;
+        let res = stream.poll_flush(writable_context).map_err(Error::io)?;
         self.socket_state.handle_write_poll(res);
         Ok(())
     }
@@ -439,7 +439,8 @@ impl<
 
         let res = self
             .send_buffer
-            .poll_write_to(writable_context, stream.as_mut())?;
+            .poll_write_to(writable_context, stream.as_mut())
+            .map_err(Error::io)?;
 
         if let Some(sz) = self.socket_state.handle_write_poll(res) {
             if sz > 0 {
@@ -497,7 +498,8 @@ impl<
             _ => {
                 let res = self
                     .receive_buffer
-                    .poll_read_from(readable_context, stream)?;
+                    .poll_read_from(readable_context, stream)
+                    .map_err(Error::io)?;
 
                 if let Some(sz) = self.socket_state.handle_read_poll(res) {
                     if sz > 0 {
@@ -538,7 +540,7 @@ impl<
             return Ok(true);
         }
         self.socket_state
-            .handle_io_result(Err(io::Error::from(io::ErrorKind::ConnectionAborted).into()))?;
+            .handle_io_result(Err(Error::io(io::ErrorKind::ConnectionAborted.into())))?;
         Ok(false)
     }