avoid runtime calls as much as possible

Author: Keruspe

src/connection.rs

@@ -9,6 +9,7 @@ use crate::{
     heartbeat::Heartbeat,
     internal_rpc::{InternalRPC, InternalRPCHandle},
     io_loop::IoLoop,
+    reactor::FullReactor,
     runtime,
     socket_state::SocketState,
     tcp::{AMQPUriTcpExt, OwnedTLSConfig},
@@ -18,6 +19,7 @@ use crate::{
 };
 use amq_protocol::frame::{AMQPFrame, ProtocolVersion};
 use async_trait::async_trait;
+use executor_trait::FullExecutor;
 use std::{fmt, io, sync::Arc};
 use tracing::{Level, level_enabled, trace};
 
@@ -181,7 +183,11 @@ impl Connection {
     pub async fn connector(
         uri: AMQPUri,
         connect: Box<
-            dyn (Fn(AMQPUri) -> Box<dyn Future<Output = io::Result<AsyncTcpStream>> + Send>)
+            dyn (Fn(
+                    AMQPUri,
+                    Arc<dyn FullExecutor + Send + Sync + 'static>,
+                    Arc<dyn FullReactor + Send + Sync + 'static>,
+                ) -> Box<dyn Future<Output = io::Result<AsyncTcpStream>> + Send>)
                 + Send
                 + Sync,
         >,
@@ -194,15 +200,15 @@ impl Connection {
         let frames = Frames::default();
         let socket_state = SocketState::default();
         let internal_rpc = InternalRPC::new(executor.clone(), socket_state.handle());
-        let heartbeat = Heartbeat::new(status.clone(), executor.clone(), reactor);
+        let heartbeat = Heartbeat::new(status.clone(), executor.clone(), reactor.clone());
         let channels = Channels::new(
             configuration.clone(),
             status.clone(),
             socket_state.handle(),
             internal_rpc.handle(),
             frames.clone(),
             heartbeat.clone(),
-            executor,
+            executor.clone(),
             uri.clone(),
             options.clone(),
         );
@@ -220,7 +226,7 @@ impl Connection {
         );
 
         internal_rpc.start(conn.channels.clone());
-        conn.io_loop.register(io_loop.start()?);
+        conn.io_loop.register(io_loop.start(executor, reactor)?);
         conn.start(uri, options).await
     }
 
@@ -288,12 +294,10 @@ impl Connect for AMQPUri {
         options: ConnectionProperties,
         config: OwnedTLSConfig,
     ) -> Result<Connection> {
-        let executor = runtime::executor()?;
-        let reactor = runtime::reactor()?;
         let config = Arc::new(config);
         Connection::connector(
             self,
-            Box::new(move |uri| {
+            Box::new(move |uri, executor, reactor| {
                 let config = config.clone();
                 let reactor = reactor.clone();
                 let executor = executor.clone();

src/io_loop.rs

@@ -9,12 +9,14 @@ use crate::{
     internal_rpc::InternalRPCHandle,
     killswitch::KillSwitch,
     protocol::{self, AMQPError, AMQPHardError},
+    reactor::FullReactor,
     socket_state::SocketState,
     thread::JoinHandle,
     types::FrameSize,
     uri::AMQPUri,
 };
 use amq_protocol::frame::{AMQPFrame, GenError, gen_frame, parse_frame};
+use executor_trait::FullExecutor;
 use futures_io::{AsyncRead, AsyncWrite};
 use std::{
     collections::VecDeque,
@@ -44,7 +46,11 @@ pub struct IoLoop {
     heartbeat: Heartbeat,
     socket_state: SocketState,
     connect: Arc<
-        dyn (Fn(AMQPUri) -> Box<dyn Future<Output = io::Result<AsyncTcpStream>> + Send>)
+        dyn (Fn(
+                AMQPUri,
+                Arc<dyn FullExecutor + Send + Sync + 'static>,
+                Arc<dyn FullReactor + Send + Sync + 'static>,
+            ) -> Box<dyn Future<Output = io::Result<AsyncTcpStream>> + Send>)
             + Send
             + Sync,
     >,
@@ -65,7 +71,11 @@ impl IoLoop {
         frames: Frames,
         socket_state: SocketState,
         connect: Arc<
-            dyn (Fn(AMQPUri) -> Box<dyn Future<Output = io::Result<AsyncTcpStream>> + Send>)
+            dyn (Fn(
+                    AMQPUri,
+                    Arc<dyn FullExecutor + Send + Sync + 'static>,
+                    Arc<dyn FullReactor + Send + Sync + 'static>,
+                ) -> Box<dyn Future<Output = io::Result<AsyncTcpStream>> + Send>)
                 + Send
                 + Sync,
         >,
@@ -166,7 +176,11 @@ impl IoLoop {
         }
     }
 
-    pub(crate) fn start(mut self) -> Result<JoinHandle> {
+    pub(crate) fn start(
+        mut self,
+        executor: Arc<dyn FullExecutor + Send + Sync + 'static>,
+        reactor: Arc<dyn FullReactor + Send + Sync + 'static>,
+    ) -> Result<JoinHandle> {
         let waker = self.socket_state.handle();
         let current_span = tracing::Span::current();
         let handle = ThreadBuilder::new()
@@ -180,7 +194,7 @@ impl IoLoop {
                 let mut writable_context = Context::from_waker(&writable_waker);
                 let (mut stream, res) = loop {
                     let (promise, resolver) = Promise::new();
-                    let connect = Box::into_pin((self.connect)(self.uri.clone()));
+                    let connect = Box::into_pin((self.connect)(self.uri.clone(), executor.clone(), reactor.clone()));
                     self.internal_rpc.register_internal_future_with_resolver(async move { Ok(connect.await?) }, resolver);
                     let mut stream = promise.wait().inspect_err(|err| {
                         trace!("Poison connection attempt");