use async-rs to check for runtime shutdown errors

Author: Keruspe

Cargo.lock

@@ -20,7 +20,7 @@ default-runtime           = ["tokio"]
 async-global-executor     = ["amq-protocol/async-global-executor", "async-rs/async-global-executor", "async-rs/async-io"]
 hickory-dns               = ["amq-protocol/hickory-dns", "async-rs/hickory-dns"]
 smol                      = ["amq-protocol/smol", "async-rs/smol"]
-tokio                     = ["amq-protocol/tokio", "async-rs/tokio", "dep:tokio"]
+tokio                     = ["amq-protocol/tokio", "async-rs/tokio"]
 
 codegen                   = ["codegen-internal", "amq-protocol/codegen"]
 codegen-internal          = ["dep:amq-protocol-codegen", "dep:serde_json"]
@@ -56,7 +56,7 @@ version = "^10.0.1"
 default-features = false
 
 [dependencies.async-rs]
-version = "^0.8"
+version = "^0.8.1"
 default-features = false
 
 [dependencies.backon]
@@ -69,12 +69,6 @@ version = "^0.12"
 default-features = false
 features = ["async"]
 
-[dependencies.tokio]
-version = "^1.50"
-default-features = false
-features = ["rt"]
-optional = true
-
 [dependencies.tracing]
 version = "^0.1"
 default-features = false

Cargo.toml

@@ -317,7 +317,7 @@ impl Connect for AMQPUri {
             async move |uri, runtime| {
                 AMQPUriTcpExt::connect_with_config_async(&uri, config.as_ref(), &runtime)
                     .await
-                    .map_err(Error::io)
+                    .map_err(|err| Error::io(err, &runtime))
             },
             options,
         )

src/connection.rs

@@ -5,7 +5,7 @@ use amq_protocol::{
     frame::{GenError, ParserError, ProtocolVersion},
     protocol::AMQPErrorKind,
 };
-use cfg_if::cfg_if;
+use async_rs::{Runtime, traits::*};
 use std::{error, fmt, io, sync::Arc};
 
 /// A std Result with a lapin::Error error type
@@ -47,8 +47,8 @@ impl Error {
         io::Error::other(error).into()
     }
 
-    pub(crate) fn io(error: io::Error) -> Self {
-        if io_error_is_runtime_shutdown(&error) {
+    pub(crate) fn io<RK: RuntimeKit>(error: io::Error, rt: &Runtime<RK>) -> Self {
+        if rt.is_runtime_shutdown_error(&error) {
             ErrorKind::RuntimeShutdownError(Arc::new(error)).into()
         } else {
             error.into()
@@ -144,18 +144,6 @@ impl Error {
     }
 }
 
-cfg_if! {
-    if #[cfg(feature = "tokio")] {
-        fn io_error_is_runtime_shutdown(err: &io::Error) -> bool {
-            tokio::runtime::is_rt_shutdown_err(err)
-        }
-    } else {
-        fn io_error_is_runtime_shutdown(_: &io::Error) -> bool {
-            false
-        }
-    }
-}
-
 impl fmt::Display for Error {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self.kind() {

src/error.rs

@@ -199,6 +199,7 @@ impl<
     }
 
     pub(crate) fn start(mut self) -> Result<JoinHandle> {
+        let runtime = self.runtime.clone();
         let waker = self.socket_state.handle();
         let connect_span = tracing::Span::current();
         let handle = ThreadBuilder::new()
@@ -270,7 +271,7 @@ impl<
                         error!(?err, "Failed to close IO stream");
                     }
                 res
-            }).map_err(Error::io)?;
+            }).map_err(|err| Error::io(err, &runtime))?;
         waker.wake();
         Ok(handle)
     }
@@ -391,7 +392,9 @@ impl<
         stream: Pin<&mut T>,
         writable_context: &mut Context<'_>,
     ) -> Result<()> {
-        let res = stream.poll_flush(writable_context).map_err(Error::io)?;
+        let res = stream
+            .poll_flush(writable_context)
+            .map_err(|err| Error::io(err, &self.runtime))?;
         self.socket_state.handle_write_poll(res);
         Ok(())
     }
@@ -440,7 +443,7 @@ impl<
         let res = self
             .send_buffer
             .poll_write_to(writable_context, stream.as_mut())
-            .map_err(Error::io)?;
+            .map_err(|err| Error::io(err, &self.runtime))?;
 
         if let Some(sz) = self.socket_state.handle_write_poll(res) {
             if sz > 0 {
@@ -499,7 +502,7 @@ impl<
                 let res = self
                     .receive_buffer
                     .poll_read_from(readable_context, stream)
-                    .map_err(Error::io)?;
+                    .map_err(|err| Error::io(err, &self.runtime))?;
 
                 if let Some(sz) = self.socket_state.handle_read_poll(res) {
                     if sz > 0 {
@@ -540,7 +543,7 @@ impl<
             return Ok(true);
         }
         self.socket_state
-            .handle_io_result(Err(Error::io(io::ErrorKind::ConnectionAborted.into())))?;
+            .handle_io_result(Err(io::Error::from(io::ErrorKind::ConnectionAborted).into()))?;
         Ok(false)
     }