switch to executor-trait

Signed-off-by: Marc-Antoine Perennou <[email protected]>

Cargo.toml

@@ -38,8 +38,8 @@ optional = true
 version = "7.0.0-alpha.4"
 default-features = false
 
-[dependencies.async-global-executor]
-version = "^2.0"
+[dependencies.async-global-executor-trait]
+version = "^1.0"
 features = ["async-io"]
 
 [dependencies.flume]
@@ -59,12 +59,14 @@ default-features = false
 async-io = "^1.0"
 async-trait = "^0.1"
 blocking = "^1.0"
+executor-trait = "^1.0"
 futures-lite = "^1.7"
 parking_lot = "^0.11"
 pinky-swear = "^5.1"
 waker-fn = "^1.1"
 
 [dev-dependencies]
+async-global-executor = "^2.0"
 serde_json = "^1.0"
 waker-fn = "^1.1"
 

src/channel.rs

@@ -9,7 +9,6 @@ use crate::{
     connection_status::{ConnectionState, ConnectionStep},
     consumer::Consumer,
     consumers::Consumers,
-    executor::Executor,
     frames::{ExpectedReply, Frames},
     id_sequence::IdSequence,
     internal_rpc::InternalRPCHandle,
@@ -27,6 +26,7 @@ use crate::{
     PromiseResolver, Result,
 };
 use amq_protocol::frame::{AMQPContentHeader, AMQPFrame};
+use executor_trait::Executor;
 use serde::{Deserialize, Serialize};
 use std::{convert::TryFrom, fmt, sync::Arc};
 use tracing::{error, info, level_enabled, trace, Level};
@@ -56,7 +56,7 @@ pub struct Channel {
     waker: SocketStateHandle,
     internal_rpc: InternalRPCHandle,
     frames: Frames,
-    executor: Arc<dyn Executor>,
+    executor: Arc<dyn Executor + Send + Sync>,
     channel_closer: Option<Arc<ChannelCloser>>,
     connection_closer: Option<Arc<ConnectionCloser>>,
 }
@@ -80,7 +80,6 @@ impl fmt::Debug for Channel {
             .field("basic_get_delivery", &self.basic_get_delivery)
             .field("returned_messages", &self.returned_messages)
             .field("frames", &self.frames)
-            .field("executor", &self.executor)
             .finish()
     }
 }
@@ -94,7 +93,7 @@ impl Channel {
         waker: SocketStateHandle,
         internal_rpc: InternalRPCHandle,
         frames: Frames,
-        executor: Arc<dyn Executor>,
+        executor: Arc<dyn Executor + Send + Sync>,
         connection_closer: Option<Arc<ConnectionCloser>>,
     ) -> Channel {
         let returned_messages = ReturnedMessages::default();

src/channels.rs

@@ -1,7 +1,6 @@
 use crate::{
     connection_closer::ConnectionCloser,
     error_handler::ErrorHandler,
-    executor::Executor,
     frames::Frames,
     id_sequence::IdSequence,
     internal_rpc::InternalRPCHandle,
@@ -14,6 +13,7 @@ use crate::{
     Error, Promise, Result,
 };
 use amq_protocol::frame::{AMQPFrame, ProtocolVersion};
+use executor_trait::Executor;
 use parking_lot::Mutex;
 use std::{collections::HashMap, fmt, sync::Arc};
 use tracing::{debug, error, level_enabled, trace, Level};
@@ -24,7 +24,7 @@ pub(crate) struct Channels {
     connection_status: ConnectionStatus,
     global_registry: Registry,
     internal_rpc: InternalRPCHandle,
-    executor: Arc<dyn Executor>,
+    executor: Arc<dyn Executor + Send + Sync>,
     frames: Frames,
     error_handler: ErrorHandler,
 }
@@ -37,7 +37,7 @@ impl Channels {
         waker: SocketStateHandle,
         internal_rpc: InternalRPCHandle,
         frames: Frames,
-        executor: Arc<dyn Executor>,
+        executor: Arc<dyn Executor + Send + Sync>,
     ) -> Self {
         Self {
             inner: Arc::new(Mutex::new(Inner::new(configuration, waker))),
@@ -285,7 +285,6 @@ impl fmt::Debug for Channels {
         }
         debug
             .field("frames", &self.frames)
-            .field("executor", &self.executor)
             .field("connection_status", &self.connection_status)
             .field("error_handler", &self.error_handler)
             .finish()
@@ -316,7 +315,7 @@ impl Inner {
         global_registry: Registry,
         internal_rpc: InternalRPCHandle,
         frames: Frames,
-        executor: Arc<dyn Executor>,
+        executor: Arc<dyn Executor + Send + Sync>,
         connection_closer: Option<Arc<ConnectionCloser>>,
     ) -> Channel {
         debug!(%id, "create channel");
@@ -341,7 +340,7 @@ impl Inner {
         global_registry: Registry,
         internal_rpc: InternalRPCHandle,
         frames: Frames,
-        executor: Arc<dyn Executor>,
+        executor: Arc<dyn Executor + Send + Sync>,
         connection_closer: Arc<ConnectionCloser>,
     ) -> Result<Channel> {
         debug!("create channel");

src/connection.rs

@@ -5,7 +5,6 @@ use crate::{
     connection_closer::ConnectionCloser,
     connection_properties::ConnectionProperties,
     connection_status::{ConnectionState, ConnectionStatus, ConnectionStep},
-    executor::{DefaultExecutor, Executor},
     frames::Frames,
     heartbeat::Heartbeat,
     internal_rpc::{InternalRPC, InternalRPCHandle},
@@ -24,6 +23,7 @@ use crate::{
 };
 use amq_protocol::frame::{AMQPFrame, ProtocolVersion};
 use async_trait::async_trait;
+use executor_trait::Executor;
 use std::{fmt, io, sync::Arc};
 use tracing::{level_enabled, Level};
 
@@ -52,7 +52,7 @@ impl Connection {
         waker: SocketStateHandle,
         internal_rpc: InternalRPCHandle,
         frames: Frames,
-        executor: Arc<dyn Executor>,
+        executor: Arc<dyn Executor + Send + Sync>,
     ) -> Self {
         let configuration = Configuration::default();
         let status = ConnectionStatus::default();
@@ -295,31 +295,37 @@ impl Connection {
         let executor = options
             .executor
             .take()
-            .map(Ok)
-            .unwrap_or_else(DefaultExecutor::default)?;
+            .unwrap_or_else(|| Arc::new(async_global_executor_trait::AsyncGlobalExecutor));
 
         let (connect_promise, resolver) = pinky_swear::PinkySwear::<Result<TcpStream>>::new();
         let connect_uri = uri.clone();
-        executor.spawn_blocking(Box::new(move || {
-            let mut res = connect(&connect_uri);
-            loop {
-                match res {
-                    Ok(stream) => {
-                        resolver.swear(Ok(stream));
-                        break;
-                    }
-                    Err(mid) => match mid.into_mid_handshake_tls_stream() {
-                        Err(err) => {
-                            resolver.swear(Err(err.into()));
-                            break;
+        executor.spawn({
+            let executor = executor.clone();
+            Box::pin(async move {
+                executor
+                    .spawn_blocking(Box::new(move || {
+                        let mut res = connect(&connect_uri);
+                        loop {
+                            match res {
+                                Ok(stream) => {
+                                    resolver.swear(Ok(stream));
+                                    break;
+                                }
+                                Err(mid) => match mid.into_mid_handshake_tls_stream() {
+                                    Err(err) => {
+                                        resolver.swear(Err(err.into()));
+                                        break;
+                                    }
+                                    Ok(mid) => {
+                                        res = mid.handshake();
+                                    }
+                                },
+                            }
                         }
-                        Ok(mid) => {
-                            res = mid.handshake();
-                        }
-                    },
-                }
-            }
-        }));
+                    }))
+                    .await;
+            })
+        });
 
         let reactor = options
             .reactor
@@ -480,7 +486,7 @@ mod tests {
         use crate::consumer::Consumer;
 
         // Bootstrap connection state to a consuming state
-        let executor = DefaultExecutor::default().unwrap();
+        let executor = Arc::new(async_global_executor_trait::AsyncGlobalExecutor);
         let socket_state = SocketState::default();
         let waker = socket_state.handle();
         let internal_rpc = InternalRPC::new(executor.clone(), waker.clone());
@@ -561,7 +567,7 @@ mod tests {
         // Bootstrap connection state to a consuming state
         let socket_state = SocketState::default();
         let waker = socket_state.handle();
-        let executor = DefaultExecutor::default().unwrap();
+        let executor = Arc::new(async_global_executor_trait::AsyncGlobalExecutor);
         let internal_rpc = InternalRPC::new(executor.clone(), waker.clone());
         let conn = Connection::new(
             waker,

src/connection_properties.rs

@@ -1,11 +1,12 @@
-use crate::{executor::Executor, reactor::Reactor, types::FieldTable};
+use crate::{reactor::Reactor, types::FieldTable};
+use executor_trait::Executor;
 use std::sync::Arc;
 
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct ConnectionProperties {
     pub locale: String,
     pub client_properties: FieldTable,
-    pub executor: Option<Arc<dyn Executor>>,
+    pub executor: Option<Arc<dyn Executor + Send + Sync>>,
     pub reactor: Option<Arc<dyn Reactor + Send + Sync>>,
 }
 
@@ -21,7 +22,7 @@ impl Default for ConnectionProperties {
 }
 
 impl ConnectionProperties {
-    pub fn with_executor<E: Executor + 'static>(mut self, executor: E) -> Self {
+    pub fn with_executor<E: Executor + Send + Sync + 'static>(mut self, executor: E) -> Self {
         self.executor = Some(Arc::new(executor));
         self
     }

src/consumer.rs

@@ -2,7 +2,6 @@ use crate::{
     channel_closer::ChannelCloser,
     consumer_canceler::ConsumerCanceler,
     consumer_status::{ConsumerState, ConsumerStatus},
-    executor::Executor,
     internal_rpc::InternalRPCHandle,
     message::{Delivery, DeliveryResult},
     options::BasicConsumeOptions,
@@ -11,6 +10,7 @@ use crate::{
     wakers::Wakers,
     BasicProperties, Error, Result,
 };
+use executor_trait::Executor;
 use flume::{Receiver, Sender};
 use futures_lite::Stream;
 use parking_lot::Mutex;
@@ -146,7 +146,7 @@ pub struct Consumer {
 impl Consumer {
     pub(crate) fn new(
         consumer_tag: ShortString,
-        executor: Arc<dyn Executor>,
+        executor: Arc<dyn Executor + Send + Sync>,
         channel_closer: Option<Arc<ChannelCloser>>,
         queue: ShortString,
         options: BasicConsumeOptions,
@@ -275,16 +275,14 @@ struct ConsumerInner {
     wakers: Wakers,
     tag: ShortString,
     delegate: Option<Arc<Box<dyn ConsumerDelegate>>>,
-    executor: Arc<dyn Executor>,
+    executor: Arc<dyn Executor + Send + Sync>,
 }
 
 impl fmt::Debug for Consumer {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let mut debug = f.debug_struct("Consumer");
         if let Some(inner) = self.inner.try_lock() {
-            debug
-                .field("tag", &inner.tag)
-                .field("executor", &inner.executor);
+            debug.field("tag", &inner.tag);
         }
         if let Some(status) = self.status.try_lock() {
             debug.field("state", &status.state());
@@ -294,7 +292,11 @@ impl fmt::Debug for Consumer {
 }
 
 impl ConsumerInner {
-    fn new(status: ConsumerStatus, consumer_tag: ShortString, executor: Arc<dyn Executor>) -> Self {
+    fn new(
+        status: ConsumerStatus,
+        consumer_tag: ShortString,
+        executor: Arc<dyn Executor + Send + Sync>,
+    ) -> Self {
         let (sender, receiver) = flume::unbounded();
         Self {
             status,
@@ -428,7 +430,6 @@ impl Stream for Consumer {
 #[cfg(test)]
 mod futures_tests {
     use super::*;
-    use crate::executor::DefaultExecutor;
 
     use std::sync::{
         atomic::{AtomicUsize, Ordering},
@@ -452,7 +453,7 @@ mod futures_tests {
 
         let mut consumer = Consumer::new(
             ShortString::from("test-consumer"),
-            DefaultExecutor::default().unwrap(),
+            Arc::new(async_global_executor_trait::AsyncGlobalExecutor),
             None,
             "test".into(),
             BasicConsumeOptions::default(),
@@ -488,7 +489,7 @@ mod futures_tests {
 
         let mut consumer = Consumer::new(
             ShortString::from("test-consumer"),
-            DefaultExecutor::default().unwrap(),
+            Arc::new(async_global_executor_trait::AsyncGlobalExecutor),
             None,
             "test".into(),
             BasicConsumeOptions::default(),