Allow running SharedConnection with any other runtime

Cargo.toml

@@ -18,16 +18,17 @@ url = "1.2"
 combine = "3.8.1"
 bytes = "0.4"
 futures = "0.1"
-tokio-executor = "0.1"
+tokio-executor = { version = "0.1", optional = true }
 tokio-tcp = "0.1"
 tokio-io = "0.1"
 tokio-codec = "0.1"
 tokio-sync = "0.1"
 
 [features]
-default = [ "geospatial" ]
+default = [ "executor", "geospatial" ]
 
 geospatial = []
+executor = [ "tokio-executor" ]
 
 [target.'cfg(unix)'.dependencies]
 tokio-uds = { version = "0.2" }

appveyor.yml

@@ -20,4 +20,4 @@ install:
 build: false
 
 test_script:
-  - cargo test --verbose --no-default-features %cargoflags%
+  - cargo test --verbose --no-default-features --features executor %cargoflags%

src/aio.rs

@@ -8,11 +8,10 @@ use std::net::ToSocketAddrs;
 use tokio_uds::UnixStream;
 
 use tokio_codec::{Decoder, Framed};
-use tokio_executor;
 use tokio_io::{self, AsyncWrite};
 use tokio_tcp::TcpStream;
 
-use futures::future::Either;
+use futures::future::{Either, Executor};
 use futures::{future, Async, AsyncSink, Future, Poll, Sink, StartSend, Stream};
 use tokio_sync::{mpsc, oneshot};
 
@@ -431,18 +430,21 @@ where
     T::Error: Send,
     T::Error: ::std::fmt::Debug,
 {
-    fn new(sink_stream: T) -> Self {
+    fn new<E>(sink_stream: T, executor: E) -> Self
+    where
+        E: Executor<Box<dyn Future<Item = (), Error = ()> + Send>>,
+    {
         const BUFFER_SIZE: usize = 50;
         let (sender, receiver) = mpsc::channel(BUFFER_SIZE);
-        tokio_executor::spawn(
-            receiver
-                .map_err(|_| ())
-                .forward(PipelineSink {
-                    sink_stream,
-                    in_flight: VecDeque::new(),
-                })
-                .map(|_| ()),
-        );
+        let f = receiver
+            .map_err(|_| ())
+            .forward(PipelineSink {
+                sink_stream,
+                in_flight: VecDeque::new(),
+            })
+            .map(|_| ());
+
+        executor.execute(Box::new(f));
         Pipeline(sender)
     }
 
@@ -498,17 +500,20 @@ pub struct SharedConnection {
 }
 
 impl SharedConnection {
-    pub fn new(con: Connection) -> impl Future<Item = Self, Error = RedisError> {
+    pub fn new<E>(con: Connection, executor: E) -> impl Future<Item = Self, Error = RedisError>
+    where
+        E: Executor<Box<dyn Future<Item = (), Error = ()> + Send>>,
+    {
         future::lazy(|| {
             let pipeline = match con.con {
                 ActualConnection::Tcp(tcp) => {
                     let codec = ValueCodec::default().framed(tcp.into_inner());
-                    ActualPipeline::Tcp(Pipeline::new(codec))
+                    ActualPipeline::Tcp(Pipeline::new(codec, executor))
                 }
                 #[cfg(unix)]
                 ActualConnection::Unix(unix) => {
                     let codec = ValueCodec::default().framed(unix.into_inner());
-                    ActualPipeline::Unix(Pipeline::new(codec))
+                    ActualPipeline::Unix(Pipeline::new(codec, executor))
                 }
             };
             Ok(SharedConnection {

src/client.rs

@@ -1,6 +1,7 @@
 use futures::Future;
 
 use connection::{connect, Connection, ConnectionInfo, ConnectionLike, IntoConnectionInfo};
+use futures::future::Executor;
 use types::{RedisError, RedisResult, Value};
 
 /// The client type.
@@ -50,11 +51,24 @@ impl Client {
         ::aio::connect(self.connection_info.clone())
     }
 
+    #[cfg(feature = "executor")]
     pub fn get_shared_async_connection(
         &self,
     ) -> impl Future<Item = ::aio::SharedConnection, Error = RedisError> {
+        let executor = tokio_executor::DefaultExecutor::current();
         self.get_async_connection()
-            .and_then(move |con| ::aio::SharedConnection::new(con))
+            .and_then(move |con| ::aio::SharedConnection::new(con, executor))
+    }
+
+    pub fn get_shared_async_connection_with_executor<E>(
+        &self,
+        executor: E,
+    ) -> impl Future<Item = ::aio::SharedConnection, Error = RedisError>
+    where
+        E: Executor<Box<dyn Future<Item = (), Error = ()> + Send>>,
+    {
+        self.get_async_connection()
+            .and_then(move |con| ::aio::SharedConnection::new(con, executor))
     }
 }
 

src/lib.rs

@@ -356,7 +356,6 @@ extern crate sha1;
 extern crate url;
 #[macro_use]
 extern crate futures;
-extern crate tokio_executor;
 #[macro_use]
 extern crate tokio_io;
 extern crate tokio_codec;