Fix connection with tokio

Recent changes moved the reactor register in the io_loop.
The io_loop thread is not managed by an executor, and tokio doesn't like
this.
This could and should be fixed in tokio-reactor-trait but requires a
breaking change there.
In the meantime, let's just move that part to the internal RPC task
which runs inside the executor.
While at it, make sure the tokio example runs properly by integrating in
the tests.

Fixes #436

Author: Keruspe

examples/tokio.rs

@@ -1,20 +1,28 @@
 use lapin::{
     BasicProperties, Connection, ConnectionProperties,
     message::DeliveryResult,
-    options::{BasicAckOptions, BasicConsumeOptions, BasicPublishOptions, QueueDeclareOptions},
+    options::{
+        BasicAckOptions, BasicCancelOptions, BasicConsumeOptions, BasicPublishOptions,
+        QueueDeclareOptions,
+    },
     types::FieldTable,
 };
 
-#[tokio::main]
-async fn main() {
-    let uri = "amqp://localhost:5672";
+async fn tokio_main(forever: bool) {
+    if std::env::var("RUST_LOG").is_err() {
+        unsafe { std::env::set_var("RUST_LOG", "info") };
+    }
+
+    tracing_subscriber::fmt::init();
+
+    let addr = std::env::var("AMQP_ADDR").unwrap_or_else(|_| "amqp://127.0.0.1:5672/%2f".into());
     let options = ConnectionProperties::default()
         // Use tokio executor and reactor.
         // At the moment the reactor is only available for unix.
         .with_executor(tokio_executor_trait::Tokio::current())
         .with_reactor(tokio_reactor_trait::Tokio);
 
-    let connection = Connection::connect(uri, options).await.unwrap();
+    let connection = Connection::connect(&addr, options).await.unwrap();
     let channel = connection.create_channel().await.unwrap();
 
     let _queue = channel
@@ -70,5 +78,25 @@ async fn main() {
         .await
         .unwrap();
 
-    std::future::pending::<()>().await;
+    if forever {
+        std::future::pending::<()>().await;
+    } else {
+        channel
+            .basic_cancel("tag_foo", BasicCancelOptions::default())
+            .await
+            .unwrap();
+        connection.close(200, "OK").await.unwrap();
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    tokio_main(true).await
+}
+
+#[test]
+fn connection() {
+    tokio::runtime::Runtime::new()
+        .expect("failed to build tokio runtime")
+        .block_on(tokio_main(false));
 }

src/connection.rs

@@ -188,8 +188,9 @@ impl Connection {
         let status = ConnectionStatus::new(&uri);
         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.clone());
+        let internal_rpc =
+            InternalRPC::new(executor.clone(), reactor.clone(), socket_state.handle());
+        let heartbeat = Heartbeat::new(status.clone(), executor.clone(), reactor);
         let channels = Channels::new(
             configuration.clone(),
             status.clone(),
@@ -215,7 +216,7 @@ impl Connection {
         );
 
         internal_rpc.start(conn.channels.clone());
-        conn.io_loop.register(io_loop.start(reactor)?);
+        conn.io_loop.register(io_loop.start()?);
         conn.start(uri, options).await
     }
 
@@ -326,7 +327,8 @@ mod tests {
         let status = ConnectionStatus::new(&uri);
         let frames = Frames::default();
         let socket_state = SocketState::default();
-        let internal_rpc = InternalRPC::new(executor.clone(), socket_state.handle());
+        let internal_rpc =
+            InternalRPC::new(executor.clone(), reactor.clone(), socket_state.handle());
         let heartbeat = Heartbeat::new(status.clone(), executor.clone(), reactor);
         let channels = Channels::new(
             configuration.clone(),

src/internal_rpc.rs

@@ -5,18 +5,21 @@ use crate::{
     error_holder::ErrorHolder,
     killswitch::KillSwitch,
     options::{BasicAckOptions, BasicCancelOptions, BasicNackOptions, BasicRejectOptions},
+    reactor::FullReactor,
     socket_state::SocketStateHandle,
     types::{ChannelId, DeliveryTag, Identifier, ReplyCode},
 };
 use executor_trait::FullExecutor;
 use flume::{Receiver, Sender};
+use reactor_trait::{AsyncIOHandle, IOHandle};
 use std::{collections::HashMap, fmt, future::Future, sync::Arc};
 use tracing::trace;
 
 pub(crate) struct InternalRPC {
     rpc: Receiver<Option<InternalCommand>>,
     handle: InternalRPCHandle,
     channels_status: HashMap<ChannelId, KillSwitch>,
+    reactor: Arc<dyn FullReactor + Send + Sync>,
 }
 
 #[derive(Clone)]
@@ -127,6 +130,14 @@ impl InternalRPCHandle {
         self.send(InternalCommand::InitConnectionShutdown(error));
     }
 
+    pub(crate) fn reactor_register(
+        &self,
+        handle: IOHandle,
+        resolver: PromiseResolver<Box<dyn AsyncIOHandle + Send>>,
+    ) {
+        self.send(InternalCommand::ReactorRegister(handle, resolver));
+    }
+
     pub(crate) fn remove_channel(&self, channel_id: ChannelId, error: Error) {
         self.send(InternalCommand::RemoveChannel(channel_id, error));
     }
@@ -230,6 +241,7 @@ enum InternalCommand {
     FinishConnectionShutdown,
     InitConnectionRecovery(Error),
     InitConnectionShutdown(Error),
+    ReactorRegister(IOHandle, PromiseResolver<Box<dyn AsyncIOHandle + Send>>),
     RemoveChannel(ChannelId, Error),
     SendConnectionCloseOk(Error),
     SetChannelStatus(ChannelId, KillSwitch),
@@ -242,6 +254,7 @@ enum InternalCommand {
 impl InternalRPC {
     pub(crate) fn new(
         executor: Arc<dyn FullExecutor + Send + Sync>,
+        reactor: Arc<dyn FullReactor + Send + Sync>,
         waker: SocketStateHandle,
     ) -> Self {
         let (sender, rpc) = flume::unbounded();
@@ -254,6 +267,7 @@ impl InternalRPC {
             rpc,
             handle,
             channels_status: Default::default(),
+            reactor,
         }
     }
 
@@ -368,6 +382,9 @@ impl InternalRPC {
                     channels.init_connection_recovery(error);
                 }
                 InitConnectionShutdown(error) => channels.init_connection_shutdown(error),
+                ReactorRegister(handle, resolver) => {
+                    resolver.complete(self.reactor.register(handle).map_err(Error::from));
+                }
                 RemoveChannel(channel_id, error) => {
                     if !self.channel_ok(channel_id) {
                         continue;

src/io_loop.rs

@@ -1,5 +1,5 @@
 use crate::{
-    Configuration, ConnectionStatus, Error, ErrorKind, PromiseResolver, Result, TcpStream,
+    Configuration, ConnectionStatus, Error, ErrorKind, Promise, PromiseResolver, Result, TcpStream,
     buffer::Buffer,
     channels::Channels,
     connection_status::ConnectionState,
@@ -8,7 +8,6 @@ use crate::{
     internal_rpc::InternalRPCHandle,
     killswitch::KillSwitch,
     protocol::{self, AMQPError, AMQPHardError},
-    reactor::FullReactor,
     socket_state::SocketState,
     tcp::HandshakeResult,
     thread::JoinHandle,
@@ -160,10 +159,7 @@ impl IoLoop {
         }
     }
 
-    pub(crate) fn start(
-        mut self,
-        reactor: Arc<dyn FullReactor + Send + Sync>,
-    ) -> Result<JoinHandle> {
+    pub(crate) fn start(mut self) -> Result<JoinHandle> {
         let waker = self.socket_state.handle();
         let current_span = tracing::Span::current();
         let handle = ThreadBuilder::new()
@@ -176,7 +172,10 @@ impl IoLoop {
                 let writable_waker = self.socket_state.writable_waker();
                 let mut writable_context = Context::from_waker(&writable_waker);
                 let (mut stream, res) = loop {
-                    let mut stream = Box::into_pin(reactor.register(IOHandle::new(self.tcp_connect()?))?);
+                    let (promise, resolver) = Promise::new();
+                    let handle = IOHandle::new(self.tcp_connect()?);
+                    self.internal_rpc.reactor_register(handle, resolver);
+                    let mut stream = Box::into_pin(promise.wait()?);
                     let mut res = Ok(());
 
                     while self.should_continue(&connection_killswitch) {

src/promise.rs

@@ -3,6 +3,7 @@ use flume::{Receiver, Sender};
 use std::{
     fmt,
     future::Future,
+    io,
     pin::Pin,
     sync::{Arc, RwLock},
     task::{Context, Poll},
@@ -65,6 +66,12 @@ impl<T: Send + 'static> Promise<T> {
     pub(crate) fn try_wait(&self) -> Option<Result<T>> {
         self.recv.try_recv().ok()
     }
+
+    pub(crate) fn wait(&self) -> Result<T> {
+        self.recv
+            .recv()
+            .unwrap_or_else(|err| Err(io::Error::other(err).into()))
+    }
 }
 
 impl<T: Send + 'static> Future for Promise<T> {

tests/tokio.rs

@@ -0,0 +1 @@
+../examples/tokio.rs