pool: use Executor trait instead of concrete type

Alternative to #24.

Currently, `Pool` cannot be used at all. It requires a concrete `Exec`,
and that is private. The #24 PR simply made `Exec` public. However,
there is already other things in this crate that require an `Exec`
(`legacy::Client`).

This adapts `Pool` to follow the same pattern as used for `Client`,
taking in a trait.

This change is NOT a breaking change, as `Pool` cannot be used at all
currently, so no one could be broken.

src/client/legacy.rs

@@ -1413,15 +1413,16 @@ impl Builder {
         B: Body + Send,
         B::Data: Send,
     {
+        let exec = self.exec.clone();
         Client {
             config: self.client_config,
-            exec: self.exec.clone(),
+            exec: exec.clone(),
             #[cfg(feature = "http1")]
             h1_builder: self.h1_builder.clone(),
             #[cfg(feature = "http2")]
             h2_builder: self.h2_builder.clone(),
             connector,
-            pool: pool::Pool::new(self.pool_config, &self.exec),
+            pool: pool::Pool::new(self.pool_config, exec),
         }
     }
 }

src/client/pool.rs

@@ -20,7 +20,7 @@ use tokio::time::{Duration, Instant, Interval};
 use futures_channel::oneshot;
 use tracing::{debug, trace};
 
-use crate::common::{exec::Exec, ready};
+use crate::common::{exec, exec::Exec, ready};
 
 // FIXME: allow() required due to `impl Trait` leaking types to this lint
 #[allow(missing_debug_implementations)]
@@ -121,7 +121,10 @@ impl Config {
 }
 
 impl<T, K: Key> Pool<T, K> {
-    pub fn new(config: Config, __exec: &Exec) -> Pool<T, K> {
+    pub fn new<E>(config: Config, executor: E) -> Pool<T, K>
+        where
+            E: hyper::rt::Executor<exec::BoxSendFuture> + Send + Sync + Clone + 'static {
+        let exec = Exec::new(executor);
         let inner = if config.is_enabled() {
             Some(Arc::new(Mutex::new(PoolInner {
                 connecting: HashSet::new(),
@@ -131,7 +134,7 @@ impl<T, K: Key> Pool<T, K> {
                 max_idle_per_host: config.max_idle_per_host,
                 waiters: HashMap::new(),
                 #[cfg(feature = "runtime")]
-                exec: __exec.clone(),
+                exec,
                 timeout: config.idle_timeout,
             })))
         } else {
@@ -830,7 +833,7 @@ mod tests {
     use std::time::Duration;
 
     use super::{Connecting, Key, Pool, Poolable, Reservation, WeakOpt};
-    use crate::common::exec::Exec;
+    use crate::rt::tokio_executor::TokioExecutor;
 
     #[derive(Clone, Debug, PartialEq, Eq, Hash)]
     struct KeyImpl(http::uri::Scheme, http::uri::Authority);
@@ -876,7 +879,7 @@ mod tests {
                 idle_timeout: Some(Duration::from_millis(100)),
                 max_idle_per_host: max_idle,
             },
-            &Exec::Default,
+            TokioExecutor::new(),
         );
         pool.no_timer();
         pool
@@ -977,7 +980,7 @@ mod tests {
                 idle_timeout: Some(Duration::from_millis(10)),
                 max_idle_per_host: std::usize::MAX,
             },
-            &Exec::Default,
+            TokioExecutor::new(),
         );
 
         let key = host_key("foo");