chore(tests): change tests to use current_thread runtime

Author: seanmonstar

Diff for: benches/end_to_end.rs

@@ -9,7 +9,7 @@ extern crate tokio;
 use std::net::SocketAddr;
 
 use futures::{Future, Stream};
-use tokio::runtime::Runtime;
+use tokio::runtime::current_thread::Runtime;
 use tokio::net::TcpListener;
 
 use hyper::{Body, Method, Request, Response};
@@ -22,22 +22,21 @@ fn get_one_at_a_time(b: &mut test::Bencher) {
     let mut rt = Runtime::new().unwrap();
     let addr = spawn_hello(&mut rt);
 
-    let connector = HttpConnector::new_with_handle(1, rt.reactor().clone());
+    let connector = HttpConnector::new(1);
     let client = hyper::Client::builder()
-        .executor(rt.executor())
         .build::<_, Body>(connector);
 
     let url: hyper::Uri = format!("http://{}/get", addr).parse().unwrap();
 
     b.bytes = 160 * 2 + PHRASE.len() as u64;
     b.iter(move || {
-        client.get(url.clone())
+        rt.block_on(client.get(url.clone())
             .and_then(|res| {
                 res.into_body().for_each(|_chunk| {
                     Ok(())
                 })
-            })
-            .wait().expect("client wait");
+            }))
+            .expect("client wait");
     });
 }
 
@@ -46,9 +45,8 @@ fn post_one_at_a_time(b: &mut test::Bencher) {
     let mut rt = Runtime::new().unwrap();
     let addr = spawn_hello(&mut rt);
 
-    let connector = HttpConnector::new_with_handle(1, rt.reactor().clone());
+    let connector = HttpConnector::new(1);
     let client = hyper::Client::builder()
-        .executor(rt.executor())
         .build::<_, Body>(connector);
 
     let url: hyper::Uri = format!("http://{}/post", addr).parse().unwrap();
@@ -59,12 +57,11 @@ fn post_one_at_a_time(b: &mut test::Bencher) {
         let mut req = Request::new(post.into());
         *req.method_mut() = Method::POST;
         *req.uri_mut() = url.clone();
-        client.request(req).and_then(|res| {
+        rt.block_on(client.request(req).and_then(|res| {
             res.into_body().for_each(|_chunk| {
                 Ok(())
             })
-        }).wait().expect("client wait");
-
+        })).expect("client wait");
     });
 }
 

Diff for: src/client/pool.rs

@@ -805,20 +805,29 @@ mod tests {
     #[cfg(feature = "runtime")]
     #[test]
     fn test_pool_timer_removes_expired() {
-        use std::sync::Arc;
-        let runtime = ::tokio::runtime::Runtime::new().unwrap();
-        let executor = runtime.executor();
-        let pool = Pool::new(true, Some(Duration::from_millis(100)), &Exec::Executor(Arc::new(executor)));
+        use std::time::Instant;
+        use tokio_timer::Delay;
+        let mut rt = ::tokio::runtime::current_thread::Runtime::new().unwrap();
+        let pool = Pool::new(true, Some(Duration::from_millis(100)), &Exec::Default);
 
         let key = (Arc::new("foo".to_string()), Ver::Http1);
 
-        pool.pooled(c(key.clone()), Uniq(41));
-        pool.pooled(c(key.clone()), Uniq(5));
-        pool.pooled(c(key.clone()), Uniq(99));
+        // Since pool.pooled() will be calling spawn on drop, need to be sure
+        // those drops are called while `rt` is the current executor. To do so,
+        // call those inside a future.
+        rt.block_on(::futures::future::lazy(|| {
+            pool.pooled(c(key.clone()), Uniq(41));
+            pool.pooled(c(key.clone()), Uniq(5));
+            pool.pooled(c(key.clone()), Uniq(99));
+            Ok::<_, ()>(())
+        })).unwrap();
 
         assert_eq!(pool.inner.connections.lock().unwrap().idle.get(&key).map(|entries| entries.len()), Some(3));
 
-        ::std::thread::sleep(Duration::from_millis(400)); // allow for too-good resolution
+        // Let the timer tick passed the expiration...
+        rt
+            .block_on(Delay::new(Instant::now() + Duration::from_millis(200)))
+            .expect("rt block_on 200ms");
 
         assert!(pool.inner.connections.lock().unwrap().idle.get(&key).is_none());
     }

Diff for: src/client/tests.rs

@@ -1,12 +1,9 @@
 #![cfg(feature = "runtime")]
 extern crate pretty_env_logger;
 
-use std::thread;
-use std::time::Duration;
-
 use futures::Async;
 use futures::future::poll_fn;
-use tokio::executor::thread_pool::{Builder as ThreadPoolBuilder};
+use tokio::runtime::current_thread::Runtime;
 
 use mock::MockConnector;
 use super::*;
@@ -15,14 +12,13 @@ use super::*;
 fn retryable_request() {
     let _ = pretty_env_logger::try_init();
 
-    let executor = ThreadPoolBuilder::new().pool_size(1).build();
+    let mut rt = Runtime::new().expect("new rt");
     let mut connector = MockConnector::new();
 
     let sock1 = connector.mock("http://mock.local");
     let sock2 = connector.mock("http://mock.local");
 
     let client = Client::builder()
-        .executor(executor.sender().clone())
         .build::<_, ::Body>(connector);
 
     client.pool.no_timer();
@@ -39,7 +35,7 @@ fn retryable_request() {
             try_ready!(sock1.write(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"));
             Ok(Async::Ready(()))
         }).map_err(|e: ::std::io::Error| panic!("srv1 poll_fn error: {}", e));
-        res1.join(srv1).wait().expect("res1");
+        rt.block_on(res1.join(srv1)).expect("res1");
     }
     drop(sock1);
 
@@ -57,20 +53,19 @@ fn retryable_request() {
         Ok(Async::Ready(()))
     }).map_err(|e: ::std::io::Error| panic!("srv2 poll_fn error: {}", e));
 
-    res2.join(srv2).wait().expect("res2");
+    rt.block_on(res2.join(srv2)).expect("res2");
 }
 
 #[test]
 fn conn_reset_after_write() {
     let _ = pretty_env_logger::try_init();
 
-    let executor = ThreadPoolBuilder::new().pool_size(1).build();
+    let mut rt = Runtime::new().expect("new rt");
     let mut connector = MockConnector::new();
 
     let sock1 = connector.mock("http://mock.local");
 
     let client = Client::builder()
-        .executor(executor.sender().clone())
         .build::<_, ::Body>(connector);
 
     client.pool.no_timer();
@@ -88,12 +83,9 @@ fn conn_reset_after_write() {
             try_ready!(sock1.write(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"));
             Ok(Async::Ready(()))
         }).map_err(|e: ::std::io::Error| panic!("srv1 poll_fn error: {}", e));
-        res1.join(srv1).wait().expect("res1");
+        rt.block_on(res1.join(srv1)).expect("res1");
     }
 
-    // sleep to allow some time for the connection to return to the pool
-    thread::sleep(Duration::from_millis(10));
-
     let req = Request::builder()
         .uri("http://mock.local/a")
         .body(Default::default())
@@ -111,9 +103,10 @@ fn conn_reset_after_write() {
         sock1.take();
         Ok(Async::Ready(()))
     }).map_err(|e: ::std::io::Error| panic!("srv2 poll_fn error: {}", e));
-    let err = res2.join(srv2).wait().expect_err("res2");
+    let err = rt.block_on(res2.join(srv2)).expect_err("res2");
     match err.kind() {
         &::error::Kind::Incomplete => (),
         other => panic!("expected Incomplete, found {:?}", other)
     }
 }
+