feat(lib): convert to use tokio 0.1

BREAKING CHANGE: All uses of `Handle` now need to be new-tokio `Handle`. Co-authored-by: Sean McArthur <[email protected]>
hyperium · Mar 19, 2018 · 27b8db3 · 27b8db3
1 parent 8c52c2d
commit 27b8db3
Show file tree

Hide file tree

Showing 23 changed files with 679 additions and 504 deletions.
diff --git a/.appveyor.yml b/.appveyor.yml
@@ -1,4 +1,5 @@
 environment:
+ RUST_BACKTRACE: 1
  matrix:
  - TARGET: x86_64-pc-windows-msvc
  - TARGET: i686-pc-windows-msvc
@@ -8,6 +9,7 @@ install:
  - curl -sSf -o rustup-init.exe https://win.rustup.rs/
  - rustup-init.exe -y --default-host %TARGET%
  - set PATH=%PATH%;C:\Users\appveyor\.cargo\bin
+
  - rustc -vV
  - cargo -vV
 build: false

diff --git a/.travis.yml b/.travis.yml
@@ -54,4 +54,5 @@ after_success:
 
 env:
  global:
+ - RUST_BACKTRACE=1
  - secure: KipdEhZsGIrb2W0HsDbC95x8FJ1RKEWPq8uSK8wSZwGw6MtvoZDX0edfrtf4o3/skA0h84yn35ZWF/rpo1ZEesgFY1g+l+me+jtyGvMwEsXTGjNP4oNR2MrDizjO8eYDm4hRUCLEmJVvsq4j7oNVdLGHfdrcnwqk8/NxJsRzqXM=
diff --git a/Cargo.toml b/Cargo.toml
@@ -25,16 +25,19 @@ base64 = "0.9"
 bytes = "0.4.4"
 futures = "0.1.17"
 futures-cpupool = "0.1.6"
+futures-timer = "0.1.0"
 http = "0.1.5"
 httparse = "1.0"
 iovec = "0.1"
 language-tags = "0.2"
 log = "0.4"
 mime = "0.3.2"
+net2 = "0.2.32"
 percent-encoding = "1.0"
 relay = "0.1"
 time = "0.1"
-tokio-core = "0.1.11"
+tokio = "0.1.3"
+tokio-executor = "0.1.0"
 tokio-service = "0.1"
 tokio-io = "0.1"
 unicase = "2.0"

diff --git a/benches/end_to_end.rs b/benches/end_to_end.rs
@@ -4,48 +4,47 @@
 extern crate futures;
 extern crate hyper;
 extern crate test;
-extern crate tokio_core;
+extern crate tokio;
 
 use std::net::SocketAddr;
 
 use futures::{Future, Stream};
-use tokio_core::reactor::{Core, Handle};
-use tokio_core::net::TcpListener;
+use tokio::runtime::Runtime;
+use tokio::net::TcpListener;
 
 use hyper::{Body, Method, Request, Response};
+use hyper::server::Http;
 
 
 #[bench]
 fn get_one_at_a_time(b: &mut test::Bencher) {
- let mut core = Core::new().unwrap();
- let handle = core.handle();
- let addr = spawn_hello(&handle);
+ let mut rt = Runtime::new().unwrap();
+ let addr = spawn_hello(&mut rt);
 
- let client = hyper::Client::new(&handle);
+ let client = hyper::Client::configure()
+ .build_with_executor(&rt.handle(), rt.executor());
 
  let url: hyper::Uri = format!("http://{}/get", addr).parse().unwrap();
 
  b.bytes = 160 * 2 + PHRASE.len() as u64;
  b.iter(move || {
- let work = client.get(url.clone()).and_then(|res| {
- res.into_body().into_stream().for_each(|_chunk| {
- Ok(())
+ client.get(url.clone())
+ .and_then(|res| {
+ res.into_body().into_stream().for_each(|_chunk| {
+ Ok(())
+ })
  })
- });
-
- core.run(work).unwrap();
+ .wait().expect("client wait");
  });
 }
 
 #[bench]
 fn post_one_at_a_time(b: &mut test::Bencher) {
- extern crate pretty_env_logger;
- let _ = pretty_env_logger::try_init();
- let mut core = Core::new().unwrap();
- let handle = core.handle();
- let addr = spawn_hello(&handle);
+ let mut rt = Runtime::new().unwrap();
+ let addr = spawn_hello(&mut rt);
 
- let client = hyper::Client::new(&handle);
+ let client = hyper::Client::configure()
+ .build_with_executor(&rt.handle(), rt.executor());
 
  let url: hyper::Uri = format!("http://{}/post", addr).parse().unwrap();
 
@@ -55,26 +54,24 @@ 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();
- let work = client.request(req).and_then(|res| {
+ client.request(req).and_then(|res| {
  res.into_body().into_stream().for_each(|_chunk| {
  Ok(())
  })
- });
+ }).wait().expect("client wait");
 
- core.run(work).unwrap();
  });
 }
 
 static PHRASE: &'static [u8] = include_bytes!("../CHANGELOG.md"); //b"Hello, World!";
 
-fn spawn_hello(handle: &Handle) -> SocketAddr {
+fn spawn_hello(rt: &mut Runtime) -> SocketAddr {
  use hyper::server::{const_service, service_fn, NewService};
  let addr = "127.0.0.1:0".parse().unwrap();
- let listener = TcpListener::bind(&addr, handle).unwrap();
+ let listener = TcpListener::bind(&addr).unwrap();
  let addr = listener.local_addr().unwrap();
 
- let handle2 = handle.clone();
- let http = hyper::server::Http::<hyper::Chunk>::new();
+ let http = Http::<hyper::Chunk>::new();
 
  let service = const_service(service_fn(|req: Request<Body>| {
  req.into_body()
@@ -85,16 +82,15 @@ fn spawn_hello(handle: &Handle) -> SocketAddr {
  })
  }));
 
- let mut conns = 0;
- handle.spawn(listener.incoming().for_each(move |(socket, _addr)| {
- conns += 1;
- assert_eq!(conns, 1, "should only need 1 connection");
- handle2.spawn(
- http.serve_connection(socket, service.new_service()?)
+ let srv = listener.incoming()
+  .into_future()
+ .map_err(|(e, _inc)| panic!("accept error: {}", e))
+ .and_then(move |(accepted, _inc)| {
+  let socket = accepted.expect("accepted socket");
+ http.serve_connection(socket, service.new_service().expect("new_service"))
  .map(|_| ())
  .map_err(|_| ())
- );
- Ok(())
- }).then(|_| Ok(())));
+ });
+ rt.spawn(srv);
  return addr
 }
diff --git a/benches/server.rs b/benches/server.rs
@@ -5,6 +5,7 @@ extern crate futures;
 extern crate hyper;
 extern crate pretty_env_logger;
 extern crate test;
+extern crate tokio;
 
 use std::io::{Read, Write};
 use std::net::{TcpListener, TcpStream};
@@ -30,7 +31,7 @@ macro_rules! bench_server {
  })).unwrap();
  let addr = srv.local_addr().unwrap();
  addr_tx.send(addr).unwrap();
- srv.run_until(until_rx.map_err(|_| ())).unwrap();
+ tokio::run(srv.run_until(until_rx.map_err(|_| ())).map_err(|e| panic!("server error: {}", e)));
  });
 
  addr_rx.recv().unwrap()

diff --git a/examples/client.rs b/examples/client.rs
@@ -1,15 +1,15 @@
 //#![deny(warnings)]
 extern crate futures;
 extern crate hyper;
-extern crate tokio_core;
+extern crate tokio;
 
 extern crate pretty_env_logger;
 
 use std::env;
 use std::io::{self, Write};
 
-use futures::Future;
-use futures::stream::Stream;
+use futures::{Future, Stream};
+use futures::future::lazy;
 
 use hyper::{Body, Client, Request};
 
@@ -30,22 +30,23 @@ fn main() {
  return;
  }
 
- let mut core = tokio_core::reactor::Core::new().unwrap();
- let handle = core.handle();
- let client = Client::new(&handle);
+ tokio::run(lazy(move || {
+ let client = Client::default();
 
- let mut req = Request::new(Body::empty());
- *req.uri_mut() = url;
- let work = client.request(req).and_then(|res| {
- println!("Response: {}", res.status());
- println!("Headers: {:#?}", res.headers());
+ let mut req = Request::new(Body::empty());
+ *req.uri_mut() = url;
 
- res.into_parts().1.into_stream().for_each(|chunk| {
- io::stdout().write_all(&chunk).map_err(From::from)
- })
- }).map(|_| {
- println!("\n\nDone.");
- });
+ client.request(req).and_then(|res| {
+ println!("Response: {}", res.status());
+ println!("Headers: {:#?}", res.headers());
 
- core.run(work).unwrap();
+ res.into_parts().1.into_stream().for_each(|chunk| {
+ io::stdout().write_all(&chunk).map_err(From::from)
+ })
+ }).map(|_| {
+ println!("\n\nDone.");
+ }).map_err(|err| {
+ eprintln!("Error {}", err);
+ })
+ }));
 }
diff --git a/examples/hello.rs b/examples/hello.rs
@@ -2,6 +2,10 @@
 extern crate hyper;
 extern crate futures;
 extern crate pretty_env_logger;
+extern crate tokio;
+
+use futures::Future;
+use futures::future::lazy;
 
 use hyper::{Body, Response};
 use hyper::server::{Http, const_service, service_fn};
@@ -16,10 +20,13 @@ fn main() {
  Ok(Response::new(Body::from(PHRASE)))
  }));
 
- let server = Http::new()
- .sleep_on_errors(true)
- .bind(&addr, new_service)
- .unwrap();
- println!("Listening on http://{} with 1 thread.", server.local_addr().unwrap());
- server.run().unwrap();
+ tokio::run(lazy(move || {
+ let server = Http::new()
+ .sleep_on_errors(true)
+ .bind(&addr, new_service)
+ .unwrap();
+
+ println!("Listening on http://{} with 1 thread.", server.local_addr().unwrap());
+ server.run().map_err(|err| eprintln!("Server error {}", err))
+ }));
 }
diff --git a/examples/multi_server.rs b/examples/multi_server.rs
@@ -1,14 +1,13 @@
 #![deny(warnings)]
 extern crate hyper;
 extern crate futures;
-extern crate tokio_core;
 extern crate pretty_env_logger;
+extern crate tokio;
 
 use futures::{Future, Stream};
-use futures::future::FutureResult;
+use futures::future::{FutureResult, lazy};
 
 use hyper::{Body, Method, Request, Response, StatusCode};
-use tokio_core::reactor::Core;
 use hyper::server::{Http, Service};
 
 static INDEX1: &'static [u8] = b"The 1st service!";
@@ -44,26 +43,23 @@ fn main() {
  let addr1 = "127.0.0.1:1337".parse().unwrap();
  let addr2 = "127.0.0.1:1338".parse().unwrap();
 
- let mut core = Core::new().unwrap();
- let handle = core.handle();
+ tokio::run(lazy(move || {
+ let srv1 = Http::new().serve_addr(&addr1, || Ok(Srv(INDEX1))).unwrap();
+ let srv2 = Http::new().serve_addr(&addr2, || Ok(Srv(INDEX2))).unwrap();
 
- let srv1 = Http::new().serve_addr_handle(&addr1, &handle, || Ok(Srv(INDEX1))).unwrap();
- let srv2 = Http::new().serve_addr_handle(&addr2, &handle, || Ok(Srv(INDEX2))).unwrap();
+  println!("Listening on http://{}", srv1.incoming_ref().local_addr());
+  println!("Listening on http://{}", srv2.incoming_ref().local_addr());
 
- println!("Listening on http://{}", srv1.incoming_ref().local_addr());
- println!("Listening on http://{}", srv2.incoming_ref().local_addr());
+ tokio::spawn(srv1.for_each(move |conn| {
+ tokio::spawn(conn.map(|_| ()).map_err(|err| println!("srv1 error: {:?}", err)));
+ Ok(())
+ }).map_err(|_| ()));
 
- let handle1 = handle.clone();
- handle.spawn(srv1.for_each(move |conn| {
- handle1.spawn(conn.map(|_| ()).map_err(|err| println!("srv1 error: {:?}", err)));
- Ok(())
- }).map_err(|_| ()));
+ tokio::spawn(srv2.for_each(move |conn| {
+ tokio::spawn(conn.map(|_| ()).map_err(|err| println!("srv2 error: {:?}", err)));
+ Ok(())
+ }).map_err(|_| ()));
 
- let handle2 = handle.clone();
- handle.spawn(srv2.for_each(move |conn| {
- handle2.spawn(conn.map(|_| ()).map_err(|err| println!("srv2 error: {:?}", err)));
  Ok(())
- }).map_err(|_| ()));
-
- core.run(futures::future::empty::<(), ()>()).unwrap();
+ }));
 }
diff --git a/examples/params.rs b/examples/params.rs
@@ -2,9 +2,11 @@
 extern crate futures;
 extern crate hyper;
 extern crate pretty_env_logger;
+extern crate tokio;
 extern crate url;
 
 use futures::{Future, Stream};
+use futures::future::lazy;
 
 use hyper::{Body, Method, Request, Response, StatusCode};
 use hyper::server::{Http, Service};
@@ -22,7 +24,7 @@ impl Service for ParamExample {
  type Request = Request<Body>;
  type Response = Response<Body>;
  type Error = hyper::Error;
- type Future = Box<Future<Item = Self::Response, Error = Self::Error>>;
+ type Future = Box<Future<Item = Self::Response, Error = Self::Error> + Send>;
 
  fn call(&self, req: Request<Body>) -> Self::Future {
  match (req.method(), req.uri().path()) {
@@ -96,7 +98,9 @@ fn main() {
  pretty_env_logger::init();
  let addr = "127.0.0.1:1337".parse().unwrap();
 
- let server = Http::new().bind(&addr, || Ok(ParamExample)).unwrap();
- println!("Listening on http://{} with 1 thread.", server.local_addr().unwrap());
- server.run().unwrap();
+ tokio::run(lazy(move || {
+ let server = Http::new().bind(&addr, || Ok(ParamExample)).unwrap();
+ println!("Listening on http://{} with 1 thread.", server.local_addr().unwrap());
+ server.run().map_err(|err| eprintln!("Server error {}", err))
+ }));
 }