Merge pull request #1482 from hyperium/unfutures-02

Revert futures 0.2 changes
hyperium · Apr 10, 2018 · 33874f9 · 33874f9
2 parents 72e02d6 + c210524
commit 33874f9
Show file tree

Hide file tree

Showing 34 changed files with 1,369 additions and 1,448 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -22,18 +22,20 @@ include = [
 
 [dependencies]
 bytes = "0.4.4"
-futures = "=0.2.0-beta"
-futures-core = "=0.2.0-beta"
-futures-timer = { git = "https://github.com/alexcrichton/futures-timer.git", rev = "682e792" }
+futures = "0.1.17"
+futures-cpupool = "0.1.6"
+futures-timer = "0.1.0"
 http = "0.1.5"
 httparse = "1.0"
 iovec = "0.1"
 log = "0.4"
 net2 = "0.2.32"
 time = "0.1"
-tokio = { git = "https://github.com/seanmonstar/tokio.git", branch = "futures2-use-after-free", features = ["unstable-futures"] }
-tokio-executor = { git = "https://github.com/seanmonstar/tokio.git", branch = "futures2-use-after-free", features = ["unstable-futures"] }
-want = { git = "https://github.com/srijs/want.git", branch = "futures-0.2" }
+tokio = "0.1.5"
+tokio-executor = "0.1.0"
+tokio-service = "0.1"
+tokio-io = "0.1"
+want = "0.0.2"
 
 [dev-dependencies]
 num_cpus = "1.0"
@@ -43,6 +45,3 @@ url = "1.0"
 
 [features]
 nightly = []
-
-[replace]
-"futures:0.2.0-beta" = { git = "https://github.com/rust-lang-nursery/futures-rs", rev = "30473ba" }
diff --git a/benches/end_to_end.rs b/benches/end_to_end.rs
@@ -8,8 +8,7 @@ extern crate tokio;
 
 use std::net::SocketAddr;
 
-use futures::{FutureExt, StreamExt};
-use futures::executor::block_on;
+use futures::{Future, Stream};
 use tokio::runtime::Runtime;
 use tokio::net::TcpListener;
 
@@ -23,20 +22,19 @@ fn get_one_at_a_time(b: &mut test::Bencher) {
     let addr = spawn_hello(&mut rt);
 
     let client = hyper::Client::configure()
-        .build(&rt.handle());
+        .build_with_executor(&rt.reactor(), rt.executor());
 
     let url: hyper::Uri = format!("http://{}/get", addr).parse().unwrap();
 
     b.bytes = 160 * 2 + PHRASE.len() as u64;
     b.iter(move || {
-        block_on(client.get(url.clone())
-            .with_executor(rt.executor())
+        client.get(url.clone())
             .and_then(|res| {
                 res.into_body().into_stream().for_each(|_chunk| {
                     Ok(())
-                }).map(|_| ())
+                })
             })
-        ).expect("client wait");
+            .wait().expect("client wait");
     });
 }
 
@@ -46,7 +44,7 @@ fn post_one_at_a_time(b: &mut test::Bencher) {
     let addr = spawn_hello(&mut rt);
 
     let client = hyper::Client::configure()
-        .build(&rt.handle());
+        .build_with_executor(&rt.reactor(), rt.executor());
 
     let url: hyper::Uri = format!("http://{}/post", addr).parse().unwrap();
 
@@ -56,14 +54,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();
-        block_on(client.request(req)
-            .with_executor(rt.executor())
-            .and_then(|res| {
-                res.into_body().into_stream().for_each(|_chunk| {
-                    Ok(())
-                }).map(|_| ())
+        client.request(req).and_then(|res| {
+            res.into_body().into_stream().for_each(|_chunk| {
+                Ok(())
             })
-        ).expect("client wait");
+        }).wait().expect("client wait");
 
     });
 }
@@ -81,22 +76,21 @@ fn spawn_hello(rt: &mut Runtime) -> SocketAddr {
     let service = const_service(service_fn(|req: Request<Body>| {
         req.into_body()
             .into_stream()
-            .concat()
+            .concat2()
             .map(|_| {
                 Response::new(Body::from(PHRASE))
             })
     }));
 
     let srv = listener.incoming()
-        .next()
+        .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(|_| ())
-        })
-        .map_err(|_| panic!("server error"));
-    rt.spawn2(srv);
+        });
+    rt.spawn(srv);
     return addr
 }
diff --git a/benches/server.rs b/benches/server.rs
@@ -11,8 +11,8 @@ use std::io::{Read, Write};
 use std::net::{TcpListener, TcpStream};
 use std::sync::mpsc;
 
-use futures::{future, stream, FutureExt, StreamExt};
-use futures::channel::oneshot;
+use futures::{future, stream, Future, Stream};
+use futures::sync::oneshot;
 
 use hyper::{Body, Request, Response};
 use hyper::server::Service;
@@ -31,7 +31,7 @@ macro_rules! bench_server {
                 })).unwrap();
                 let addr = srv.local_addr().unwrap();
                 addr_tx.send(addr).unwrap();
-                tokio::runtime::run2(srv.run_until(until_rx.map_err(|_| ())).map_err(|e| panic!("server error: {}", e)));
+                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
@@ -8,7 +8,7 @@ extern crate pretty_env_logger;
 use std::env;
 use std::io::{self, Write};
 
-use futures::{FutureExt, StreamExt};
+use futures::{Future, Stream};
 use futures::future::lazy;
 
 use hyper::{Body, Client, Request};
@@ -30,7 +30,7 @@ fn main() {
         return;
     }
 
-    tokio::runtime::run2(lazy(move |_| {
+    tokio::run(lazy(move || {
         let client = Client::default();
 
         let mut req = Request::new(Body::empty());
@@ -43,13 +43,10 @@ fn main() {
             res.into_parts().1.into_stream().for_each(|chunk| {
                 io::stdout().write_all(&chunk).map_err(From::from)
             })
-        }).then(|result| {
-            if let Some(err) = result.err() {
-                eprintln!("Error {}", err);
-            } else {
-                println!("\n\nDone.");
-            }
-            Ok(())
+        }).map(|_| {
+            println!("\n\nDone.");
+        }).map_err(|err| {
+            eprintln!("Error {}", err);
         })
     }));
 }
diff --git a/examples/hello.rs b/examples/hello.rs
@@ -4,7 +4,7 @@ extern crate futures;
 extern crate pretty_env_logger;
 extern crate tokio;
 
-use futures::FutureExt;
+use futures::Future;
 use futures::future::lazy;
 
 use hyper::{Body, Response};
@@ -20,13 +20,13 @@ fn main() {
         Ok(Response::new(Body::from(PHRASE)))
     }));
 
-    tokio::runtime::run2(lazy(move |_| {
+    tokio::run(lazy(move || {
         let server = Http::new()
             .sleep_on_errors(true)
             .bind(&addr, new_service)
             .unwrap();
 
-        println!("Listening on http://{}", server.local_addr().unwrap());
-        server.run().map_err(|err| panic!("Server error {}", err))
+        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
@@ -4,9 +4,8 @@ extern crate futures;
 extern crate pretty_env_logger;
 extern crate tokio;
 
-use futures::{FutureExt, StreamExt};
+use futures::{Future, Stream};
 use futures::future::{FutureResult, lazy};
-use futures::executor::spawn;
 
 use hyper::{Body, Method, Request, Response, StatusCode};
 use hyper::server::{Http, Service};
@@ -44,20 +43,22 @@ fn main() {
     let addr1 = "127.0.0.1:1337".parse().unwrap();
     let addr2 = "127.0.0.1:1338".parse().unwrap();
 
-    tokio::runtime::run2(lazy(move |_| {
+    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();
 
         println!("Listening on http://{}", srv1.incoming_ref().local_addr());
         println!("Listening on http://{}", srv2.incoming_ref().local_addr());
 
-        spawn(srv1.map_err(|err| panic!("srv1 error: {:?}", err)).for_each(move |conn| {
-            spawn(conn.map(|_| ()).map_err(|err| panic!("srv1 error: {:?}", err)))
-        }).map(|_| ()));
+        tokio::spawn(srv1.for_each(move |conn| {
+            tokio::spawn(conn.map(|_| ()).map_err(|err| println!("srv1 error: {:?}", err)));
+            Ok(())
+        }).map_err(|_| ()));
 
-        spawn(srv2.map_err(|err| panic!("srv2 error: {:?}", err)).for_each(move |conn| {
-            spawn(conn.map(|_| ()).map_err(|err| panic!("srv2 error: {:?}", err)))
-        }).map(|_| ()));
+        tokio::spawn(srv2.for_each(move |conn| {
+            tokio::spawn(conn.map(|_| ()).map_err(|err| println!("srv2 error: {:?}", err)));
+            Ok(())
+        }).map_err(|_| ()));
 
         Ok(())
     }));

diff --git a/examples/params.rs b/examples/params.rs
@@ -5,7 +5,7 @@ extern crate pretty_env_logger;
 extern crate tokio;
 extern crate url;
 
-use futures::{Future, FutureExt, StreamExt};
+use futures::{Future, Stream};
 use futures::future::lazy;
 
 use hyper::{Body, Method, Request, Response, StatusCode};
@@ -32,7 +32,7 @@ impl Service for ParamExample {
                 Box::new(futures::future::ok(Response::new(INDEX.into())))
             },
             (&Method::POST, "/post") => {
-                Box::new(req.into_parts().1.into_stream().concat().map(|b| {
+                Box::new(req.into_parts().1.into_stream().concat2().map(|b| {
                     // Parse the request body. form_urlencoded::parse
                     // always succeeds, but in general parsing may
                     // fail (for example, an invalid post of json), so
@@ -98,11 +98,9 @@ fn main() {
     pretty_env_logger::init();
     let addr = "127.0.0.1:1337".parse().unwrap();
 
-    tokio::runtime::run2(lazy(move |_| {
+    tokio::run(lazy(move || {
         let server = Http::new().bind(&addr, || Ok(ParamExample)).unwrap();
-        println!("Listening on http://{}", server.local_addr().unwrap());
-        server.run().recover(|err| {
-            eprintln!("Server error {}", err)
-        })
+        println!("Listening on http://{} with 1 thread.", server.local_addr().unwrap());
+        server.run().map_err(|err| eprintln!("Server error {}", err))
     }));
 }
diff --git a/examples/send_file.rs b/examples/send_file.rs
@@ -4,9 +4,9 @@ extern crate hyper;
 extern crate pretty_env_logger;
 extern crate tokio;
 
-use futures::{Future, FutureExt};
+use futures::{Future/*, Sink*/};
 use futures::future::lazy;
-use futures::channel::oneshot;
+use futures::sync::oneshot;
 
 use hyper::{Body, /*Chunk,*/ Method, Request, Response, StatusCode};
 use hyper::error::Error;
@@ -141,9 +141,9 @@ fn main() {
     pretty_env_logger::init();
     let addr = "127.0.0.1:1337".parse().unwrap();
 
-    tokio::runtime::run2(lazy(move |_| {
+    tokio::run(lazy(move || {
         let server = Http::new().bind(&addr, || Ok(ResponseExamples)).unwrap();
-        println!("Listening on http://{}", server.local_addr().unwrap());
-        server.run().map_err(|err| panic!("Server error {}", err))
+        println!("Listening on http://{} with 1 thread.", server.local_addr().unwrap());
+        server.run().map_err(|err| eprintln!("Server error {}", err))
     }));
 }
diff --git a/examples/server.rs b/examples/server.rs
@@ -4,7 +4,7 @@ extern crate hyper;
 extern crate pretty_env_logger;
 extern crate tokio;
 
-use futures::FutureExt;
+use futures::Future;
 use futures::future::{FutureResult, lazy};
 
 use hyper::{Body, Method, Request, Response, StatusCode};
@@ -43,11 +43,9 @@ fn main() {
     pretty_env_logger::init();
     let addr = "127.0.0.1:1337".parse().unwrap();
 
-    tokio::runtime::run2(lazy(move |_| {
+    tokio::run(lazy(move || {
         let server = Http::new().bind(&addr, || Ok(Echo)).unwrap();
-        println!("Listening on http://{}", server.local_addr().unwrap());
-        server.run().recover(|err| {
-            eprintln!("Server error {}", err)
-        })
+        println!("Listening on http://{} with 1 thread.", server.local_addr().unwrap());
+        server.run().map_err(|err| eprintln!("Server error {}", err))
     }));
 }
diff --git a/examples/web_api.rs b/examples/web_api.rs
@@ -4,8 +4,7 @@ extern crate hyper;
 extern crate pretty_env_logger;
 extern crate tokio;
 
-use futures::{Future, FutureExt, StreamExt};
-use futures::executor::spawn;
+use futures::{Future, Stream};
 use futures::future::lazy;
 use tokio::reactor::Handle;
 
@@ -79,15 +78,13 @@ fn main() {
     pretty_env_logger::init();
     let addr = "127.0.0.1:1337".parse().unwrap();
 
-    tokio::runtime::run2(lazy(move |_| {
+    tokio::run(lazy(move || {
         let handle = Handle::current();
         let serve = Http::new().serve_addr(&addr, move || Ok(ResponseExamples(handle.clone()))).unwrap();
-        println!("Listening on http://{}", serve.incoming_ref().local_addr());
+        println!("Listening on http://{} with 1 thread.", serve.incoming_ref().local_addr());
 
-        serve.map_err(|err| panic!("server error {:?}", err)).for_each(move |conn| {
-            spawn(conn.recover(|err| {
-                println!("connection error: {:?}", err);
-            }))
-        }).map(|_| ())
+        serve.map_err(|_| ()).for_each(move |conn| {
+            tokio::spawn(conn.map(|_| ()).map_err(|err| println!("serve error: {:?}", err)))
+        })
     }));
 }