forked/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(tests): update to newest tokio-timer

Browse Source
This commit is contained in:
Sean McArthur
2019-08-29 13:50:55 -07:00
parent 5b1feb8a3c
commit 2664cf505f
2 changed files with 25 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
tests/client.rs
View File

@@ -25,7 +25,7 @@ fn s(buf: &[u8]) -> &str {
} }
fn tcp_connect(addr: &SocketAddr) -> impl Future<Output = std::io::Result<TcpStream>> { fn tcp_connect(addr: &SocketAddr) -> impl Future<Output = std::io::Result<TcpStream>> {
TcpStream::connect(addr) TcpStream::connect(*addr)
} }
macro_rules! test { macro_rules! test {
@@ -756,7 +756,7 @@ mod dispatch_impl {
use std::sync::Arc; use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread; use std::thread;
use std::time::{Duration, Instant}; use std::time::{Duration};
use futures_core::{self, Future}; use futures_core::{self, Future};
use futures_channel::{mpsc, oneshot}; use futures_channel::{mpsc, oneshot};
@@ -767,7 +767,6 @@ mod dispatch_impl {
use tokio::runtime::current_thread::Runtime; use tokio::runtime::current_thread::Runtime;
use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::{AsyncRead, AsyncWrite};
use tokio_net::tcp::TcpStream; use tokio_net::tcp::TcpStream;
use tokio_timer::Delay;
use hyper::client::connect::{Connect, Connected, Destination, HttpConnector}; use hyper::client::connect::{Connect, Connected, Destination, HttpConnector};
use hyper::Client; use hyper::Client;
@@ -804,7 +803,7 @@ mod dispatch_impl {
.unwrap(); .unwrap();
let res = client.request(req).map_ok(move |res| { let res = client.request(req).map_ok(move |res| {
assert_eq!(res.status(), hyper::StatusCode::OK); assert_eq!(res.status(), hyper::StatusCode::OK);
Delay::new(Instant::now() + Duration::from_secs(1)) tokio_timer::sleep(Duration::from_secs(1))
}); });
let rx = rx1.expect("thread panicked"); let rx = rx1.expect("thread panicked");
rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap(); rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap();
@@ -848,7 +847,7 @@ mod dispatch_impl {
assert_eq!(res.status(), hyper::StatusCode::OK); assert_eq!(res.status(), hyper::StatusCode::OK);
res.into_body().try_concat() res.into_body().try_concat()
}).map_ok(|_| { }).map_ok(|_| {
Delay::new(Instant::now() + Duration::from_secs(1)) tokio_timer::sleep(Duration::from_secs(1))
}) })
}; };
// client is dropped // client is dropped
@@ -912,7 +911,7 @@ mod dispatch_impl {
} }
drop(client); drop(client);
let t = Delay::new(Instant::now() + Duration::from_millis(100)) let t = tokio_timer::sleep(Duration::from_millis(100))
.map(|_| panic!("time out")); .map(|_| panic!("time out"));
let close = closes let close = closes
.into_future() .into_future()
@@ -961,7 +960,7 @@ mod dispatch_impl {
rt.block_on(future::select(res, rx1)); rt.block_on(future::select(res, rx1));
// res now dropped // res now dropped
let t = Delay::new(Instant::now() + Duration::from_millis(100)) let t = tokio_timer::sleep(Duration::from_millis(100))
.map(|_| panic!("time out")); .map(|_| panic!("time out"));
let close = closes let close = closes
.into_future() .into_future()
@@ -1010,7 +1009,7 @@ mod dispatch_impl {
let rx = rx1.expect("thread panicked"); let rx = rx1.expect("thread panicked");
rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap(); rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap();
let t = Delay::new(Instant::now() + Duration::from_millis(100)) let t = tokio_timer::sleep(Duration::from_millis(100))
.map(|_| panic!("time out")); .map(|_| panic!("time out"));
let close = closes let close = closes
.into_future() .into_future()
@@ -1060,7 +1059,7 @@ mod dispatch_impl {
let rx = rx1.expect("thread panicked"); let rx = rx1.expect("thread panicked");
rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap(); rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap();
let t = Delay::new(Instant::now() + Duration::from_millis(100)) let t = tokio_timer::sleep(Duration::from_millis(100))
.map(|_| panic!("time out")); .map(|_| panic!("time out"));
let close = closes let close = closes
.into_future() .into_future()
@@ -1104,7 +1103,7 @@ mod dispatch_impl {
let rx = rx1.expect("thread panicked"); let rx = rx1.expect("thread panicked");
rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap(); rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap();
let t = Delay::new(Instant::now() + Duration::from_millis(100)) let t = tokio_timer::sleep(Duration::from_millis(100))
.map(|_| panic!("time out")); .map(|_| panic!("time out"));
let close = closes let close = closes
.into_future() .into_future()
@@ -1259,7 +1258,6 @@ mod dispatch_impl {
#[test] #[test]
fn client_keep_alive_when_response_before_request_body_ends() { fn client_keep_alive_when_response_before_request_body_ends() {
use tokio_timer::Delay;
let _ = pretty_env_logger::try_init(); let _ = pretty_env_logger::try_init();
let server = TcpListener::bind("127.0.0.1:0").unwrap(); let server = TcpListener::bind("127.0.0.1:0").unwrap();
@@ -1300,7 +1298,7 @@ mod dispatch_impl {
assert_eq!(connects.load(Ordering::Relaxed), 0); assert_eq!(connects.load(Ordering::Relaxed), 0);
let delayed_body = rx1 let delayed_body = rx1
.then(|_| Delay::new(Instant::now() + Duration::from_millis(200))) .then(|_| tokio_timer::sleep(Duration::from_millis(200)))
.map(|_| Ok::<_, ()>("hello a")) .map(|_| Ok::<_, ()>("hello a"))
.map_err(|_| -> hyper::Error { panic!("rx1") }) .map_err(|_| -> hyper::Error { panic!("rx1") })
.into_stream(); .into_stream();
@@ -1315,7 +1313,7 @@ mod dispatch_impl {
// req 1 // req 1
let fut = future::join(client.request(req), rx) let fut = future::join(client.request(req), rx)
.then(|_| Delay::new(Instant::now() + Duration::from_millis(200))) .then(|_| tokio_timer::sleep(Duration::from_millis(200)))
// req 2 // req 2
.then(move |()| { .then(move |()| {
let rx = rx3.expect("thread panicked"); let rx = rx3.expect("thread panicked");
@@ -1477,7 +1475,7 @@ mod dispatch_impl {
let _ = pretty_env_logger::try_init(); let _ = pretty_env_logger::try_init();
let mut rt = Runtime::new().unwrap(); let mut rt = Runtime::new().unwrap();
let mut listener = TcpListener::bind(&"127.0.0.1:0".parse().unwrap()).unwrap(); let mut listener = rt.block_on(TcpListener::bind("127.0.0.1:0")).unwrap();
let addr = listener.local_addr().unwrap(); let addr = listener.local_addr().unwrap();
let mut connector = DebugConnector::new(); let mut connector = DebugConnector::new();
connector.alpn_h2 = true; connector.alpn_h2 = true;
@@ -1614,7 +1612,7 @@ mod conn {
use std::pin::Pin; use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use std::thread; use std::thread;
use std::time::{Duration, Instant}; use std::time::{Duration};
use futures_channel::oneshot; use futures_channel::oneshot;
use futures_util::future::{self, poll_fn, FutureExt}; use futures_util::future::{self, poll_fn, FutureExt};
@@ -1623,7 +1621,6 @@ mod conn {
use tokio::runtime::current_thread::Runtime; use tokio::runtime::current_thread::Runtime;
use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::{AsyncRead, AsyncWrite};
use tokio_net::tcp::TcpStream; use tokio_net::tcp::TcpStream;
use tokio_timer::Delay;
use hyper::{self, Request, Body, Method}; use hyper::{self, Request, Body, Method};
use hyper::client::conn; use hyper::client::conn;
@@ -1670,7 +1667,7 @@ mod conn {
res.into_body().try_concat() res.into_body().try_concat()
}); });
let rx = rx1.expect("thread panicked"); let rx = rx1.expect("thread panicked");
let rx = rx.then(|_| Delay::new(Instant::now() + Duration::from_millis(200))); let rx = rx.then(|_| tokio_timer::sleep(Duration::from_millis(200)));
rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap(); rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap();
} }
@@ -1716,7 +1713,7 @@ mod conn {
}); });
let rx = rx1.expect("thread panicked"); let rx = rx1.expect("thread panicked");
let rx = rx.then(|_| Delay::new(Instant::now() + Duration::from_millis(200))); let rx = rx.then(|_| tokio_timer::sleep(Duration::from_millis(200)));
let chunk = rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap(); let chunk = rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap();
assert_eq!(chunk.len(), 5); assert_eq!(chunk.len(), 5);
} }
@@ -1808,7 +1805,7 @@ mod conn {
res.into_body().try_concat() res.into_body().try_concat()
}); });
let rx = rx1.expect("thread panicked"); let rx = rx1.expect("thread panicked");
let rx = rx.then(|_| Delay::new(Instant::now() + Duration::from_millis(200))); let rx = rx.then(|_| tokio_timer::sleep(Duration::from_millis(200)));
rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap(); rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap();
} }
@@ -1852,7 +1849,7 @@ mod conn {
res.into_body().try_concat() res.into_body().try_concat()
}); });
let rx = rx1.expect("thread panicked"); let rx = rx1.expect("thread panicked");
let rx = rx.then(|_| Delay::new(Instant::now() + Duration::from_millis(200))); let rx = rx.then(|_| tokio_timer::sleep(Duration::from_millis(200)));
rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap(); rt.block_on(future::join(res, rx).map(|r| r.0)).unwrap();
} }
@@ -1904,7 +1901,7 @@ mod conn {
}); });
let rx = rx1.expect("thread panicked"); let rx = rx1.expect("thread panicked");
let rx = rx.then(|_| Delay::new(Instant::now() + Duration::from_millis(200))); let rx = rx.then(|_| tokio_timer::sleep(Duration::from_millis(200)));
rt.block_on(future::join3(res1, res2, rx).map(|r| r.0)).unwrap(); rt.block_on(future::join3(res1, res2, rx).map(|r| r.0)).unwrap();
} }
@@ -1964,7 +1961,7 @@ mod conn {
}); });
let rx = rx1.expect("thread panicked"); let rx = rx1.expect("thread panicked");
let rx = rx.then(|_| Delay::new(Instant::now() + Duration::from_millis(200))); let rx = rx.then(|_| tokio_timer::sleep(Duration::from_millis(200)));
rt.block_on(future::join3(until_upgrade, res, rx).map(|r| r.0)).unwrap(); rt.block_on(future::join3(until_upgrade, res, rx).map(|r| r.0)).unwrap();
// should not be ready now // should not be ready now
@@ -2052,7 +2049,7 @@ mod conn {
}); });
let rx = rx1.expect("thread panicked"); let rx = rx1.expect("thread panicked");
let rx = rx.then(|_| Delay::new(Instant::now() + Duration::from_millis(200))); let rx = rx.then(|_| tokio_timer::sleep(Duration::from_millis(200)));
rt.block_on(future::join3(until_tunneled, res, rx).map(|r| r.0)).unwrap(); rt.block_on(future::join3(until_tunneled, res, rx).map(|r| r.0)).unwrap();
// should not be ready now // should not be ready now
@@ -2086,7 +2083,6 @@ mod conn {
use futures_util::future; use futures_util::future;
use hyper::{Response, Server}; use hyper::{Response, Server};
use hyper::service::{make_service_fn, service_fn}; use hyper::service::{make_service_fn, service_fn};
use tokio::timer::Delay;
let _ = pretty_env_logger::try_init(); let _ = pretty_env_logger::try_init();
@@ -2129,7 +2125,7 @@ mod conn {
let _ = shdn_tx.send(()); let _ = shdn_tx.send(());
// Allow time for graceful shutdown roundtrips... // Allow time for graceful shutdown roundtrips...
rt.block_on(Delay::new(::std::time::Instant::now() + Duration::from_millis(100))); rt.block_on(tokio_timer::sleep(Duration::from_millis(100)));
// After graceful shutdown roundtrips, the client should be closed... // After graceful shutdown roundtrips, the client should be closed...
rt.block_on(future::poll_fn(|ctx| client.poll_ready(ctx))).expect_err("client should be closed"); rt.block_on(future::poll_fn(|ctx| client.poll_ready(ctx))).expect_err("client should be closed");

9
tests/server.rs
View File

@@ -11,7 +11,7 @@ use std::net::{TcpListener as StdTcpListener};
use std::pin::Pin; use std::pin::Pin;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use std::thread; use std::thread;
use std::time::{Duration, Instant}; use std::time::Duration;
use futures_channel::oneshot; use futures_channel::oneshot;
use futures_core::ready; use futures_core::ready;
@@ -25,7 +25,6 @@ use tokio_net::driver::Handle;
use tokio_net::tcp::{TcpListener, TcpStream as TkTcpStream}; use tokio_net::tcp::{TcpListener, TcpStream as TkTcpStream};
use tokio::runtime::current_thread::Runtime; use tokio::runtime::current_thread::Runtime;
use tokio_io::{AsyncRead, AsyncWrite}; use tokio_io::{AsyncRead, AsyncWrite};
use tokio_timer::Delay;
use hyper::{Body, Request, Response, StatusCode, Version}; use hyper::{Body, Request, Response, StatusCode, Version};
use hyper::client::Client; use hyper::client::Client;
@@ -938,7 +937,7 @@ fn disable_keep_alive_post_request() {
// the read-blocked socket. // the read-blocked socket.
// //
// See https://github.com/carllerche/mio/issues/776 // See https://github.com/carllerche/mio/issues/776
let timeout = Delay::new(Instant::now() + Duration::from_millis(10)); let timeout = tokio_timer::sleep(Duration::from_millis(10));
rt.block_on(timeout); rt.block_on(timeout);
assert!(dropped.load()); assert!(dropped.load());
child.join().unwrap(); child.join().unwrap();
@@ -1007,7 +1006,7 @@ fn socket_half_closed() {
.map_err(|_| unreachable!()) .map_err(|_| unreachable!())
.and_then(|socket| { .and_then(|socket| {
Http::new().serve_connection(socket, service_fn(|_| { Http::new().serve_connection(socket, service_fn(|_| {
Delay::new(Instant::now() + Duration::from_millis(500)) tokio_timer::sleep(Duration::from_millis(500))
.map(|_| Ok::<_, hyper::Error>(Response::new(Body::empty()))) .map(|_| Ok::<_, hyper::Error>(Response::new(Body::empty())))
})) }))
}); });
@@ -1035,7 +1034,7 @@ fn disconnect_after_reading_request_before_responding() {
Http::new() Http::new()
.http1_half_close(false) .http1_half_close(false)
.serve_connection(socket, service_fn(|_| { .serve_connection(socket, service_fn(|_| {
Delay::new(Instant::now() + Duration::from_secs(2)) tokio_timer::sleep(Duration::from_secs(2))
.map(|_| -> Result<Response<Body>, hyper::Error> { .map(|_| -> Result<Response<Body>, hyper::Error> {
panic!("response future should have been dropped"); panic!("response future should have been dropped");
}) })