Migrate Gotham to std Futures to support Async/Await (#370)

* Ported examples to Std::Future

The following examples were not updated (and thus commented out in `Cargo.toml`) due to incompatible dependencies:

- [ ] `hello_world_until` is dependent on a compatible version of `tokio-signal`
- [ ] `diesel` is dependent on the Diesel middleware
- [ ] `openssl` is dependent on `tokio-openssl`
- [ ] `websocket` is dependent on `tokio-tungstenite`

* Migrate Gotham to std Futures to support Async/Await

With the anticipated release of async-await in stable rust this week, I took an effort to migrate Gotham to run on std futures using the pre-release versions of the dependencies (tokio, hyper, etc).

*NOTE*: This doesn't attempt to introduce `await` or `async fn` into the codebase (there was a single unavoidable case due to an `tokio::File::metadata` API change).  That is designed as a future activity.

This migration involved a few key efforts:

1. Convert from Futures 0.1 to Futures-preview 0.3 (mostly `Future<Item=>` to `Future<Output=Result<>>`).
2. Update dependencies to pre-release versions (`tokio-0.2` and `hyper-0.13`).  There's still other dependencies that are outstanding and blocking the full release.
3. Migrate Handler trait to a pinned box HandlerFuture.

This is a **work-in-progress** with a few blockers before this would be ready:

Gotham Dependencies:

- [ ] Update Futures from `futures-preview` to `futures = 0.3` when the other dependencies (hyper, tokio, etc) update in concert.
- [ ] Update Tokio to `0.2` from alpha pre-releases
- [ ] Update Hyper to `0.13` from alpha pre-releases
- [ ] Update Tower-Service to `0.3` from alpha pre-releases.  Hyper is migrating many of its traits to `tower-service::Service` and so is now a direct dependency.
- [ ] Released version of `futures_rustls` which is currently a branch of `tokio-rustls` ported to Futures-preview
- [ ] Released version of `futures-tokio-compat` or suggested `tokio::compat` library for bridging `futures::AsyncRead` and `tokio::AsyncRead`.  See tokio-rs/tokio#1297
    and async-rs/async-std#54

Middleware Dependencies:

- [ ] Diesel - Requires updated release of `tokio-threadpool`
- [ ] JWT - Requires updated release of `jsonwebtoken` since it's dependency `ring` conflicts withs `futures-rustls`.

* Updated to crates.io verison of futures-rustls

* Migrated to tokio-0.2 final and hyper-0.13 final

This involved changing the way that the runtime is managed and handling
`block_on` for synchronously running futures.

Hyper-0.13 required changing to the tower_service trait for implementing
a `Connect` trait used for the TLS and plain tcp stream wrappers.

The other major change is that Hyper migrated from a `Chunk` trait to
use `Bytes`.  Bytes-0.5 is now a trait rather than a struct which no
longer implements `Extends<u8>` requiring a new `BytesMut` buffer for
some common concatenation operations.

* Re-enable hello_world_until

Replace tokio-signal with tokio (signal was migrated into tokio itself)

* Remove commented out code

Co-Authored-By: Pavan Kumar Sunkara <pavan.sss1991@gmail.com>

* Async-ified the init_server helpers

* Simplify executing futures in tls/test to match plain/test

* Re-enabled diesel middleware and example

* Addressed review comments.

Co-authored-by: Pavan Kumar Sunkara <pavan.sss1991@gmail.com>

Cargo.toml

@@ -8,7 +8,8 @@ members = [
     ## Middleware
     "middleware/template",
     "middleware/diesel",
-    "middleware/jwt",
+    # TODO: Re-enable middleware when their dependencies are updated
+    # "middleware/jwt",
 
     ## Examples (these crates are not published)
     "examples/hello_world",
@@ -68,12 +69,14 @@ members = [
     "examples/diesel",
 
     # openssl
-    "examples/openssl",
+    # TODO: Re-enable when this example is updated
+    # "examples/openssl",
 
     # example_contribution_template
     "examples/example_contribution_template/name",
 
-    "examples/websocket",
+    # TODO: Re-enable when tokio-tungstenite is updated
+    # "examples/websocket",
 ]
 
 [patch.crates-io]

examples/cookies/introduction/Cargo.toml

@@ -8,6 +8,6 @@ publish = false
 [dependencies]
 gotham = { path = "../../../gotham" }
 
-hyper = "0.12"
+hyper = "0.13.1"
 mime = "0.3"
 cookie = "0.12"

examples/diesel/Cargo.toml

@@ -11,8 +11,9 @@ publish = false
 gotham = { path = "../../gotham/"}
 gotham_derive = { path = "../../gotham_derive/" }
 gotham_middleware_diesel = { path = "../../middleware/diesel"}
-hyper = "0.12"
-futures = "0.1"
+hyper = "0.13.1"
+failure = "0.1"
+futures = "0.3.1"
 mime = "0.3"
 log = "0.4"
 diesel = { version = "1.4", features = ["sqlite", "extras"] }
@@ -25,4 +26,4 @@ serde_derive = "1.0"
 
 [dev-dependencies]
 diesel_migrations = "1.4.0"
-tokio = "0.1"
+tokio = { version = "0.2.6", features = ["full"] }

examples/diesel/src/main.rs

@@ -9,15 +9,16 @@ extern crate diesel_migrations;
 
 use diesel::prelude::*;
 use diesel::sqlite::SqliteConnection;
-use futures::{future, Future, Stream};
+use futures::prelude::*;
 use gotham::handler::{HandlerError, HandlerFuture, IntoHandlerError};
 use gotham::helpers::http::response::create_response;
 use gotham::pipeline::{new_pipeline, single::single_pipeline};
 use gotham::router::{builder::*, Router};
 use gotham::state::{FromState, State};
 use gotham_middleware_diesel::DieselMiddleware;
-use hyper::{Body, StatusCode};
+use hyper::{body, Body, StatusCode};
 use serde_derive::Serialize;
+use std::pin::Pin;
 use std::str::from_utf8;
 
 mod models;
@@ -43,46 +44,51 @@ struct RowsUpdated {
     rows: usize,
 }
 
-fn create_product_handler(mut state: State) -> Box<HandlerFuture> {
+fn create_product_handler(mut state: State) -> Pin<Box<HandlerFuture>> {
     let repo = Repo::borrow_from(&state).clone();
-    let f = extract_json::<NewProduct>(&mut state)
-        .and_then(move |product| {
-            repo.run(move |conn| {
-                // Insert the `NewProduct` in the DB
+    async move {
+        let product = match extract_json::<NewProduct>(&mut state).await {
+            Ok(product) => product,
+            Err(e) => return Err((state, e)),
+        };
+
+        let query_result = repo
+            .run(move |conn| {
                 diesel::insert_into(products::table)
                     .values(&product)
                     .execute(&conn)
             })
-            .map_err(|e| e.into_handler_error())
-        })
-        .then(|result| match result {
-            Ok(rows) => {
-                let body = serde_json::to_string(&RowsUpdated { rows })
-                    .expect("Failed to serialise to json");
-                let res =
-                    create_response(&state, StatusCode::CREATED, mime::APPLICATION_JSON, body);
-                future::ok((state, res))
-            }
-            Err(e) => future::err((state, e)),
-        });
-    Box::new(f)
+            .await;
+
+        let rows = match query_result {
+            Ok(rows) => rows,
+            Err(e) => return Err((state, e.into_handler_error())),
+        };
+
+        let body =
+            serde_json::to_string(&RowsUpdated { rows }).expect("Failed to serialise to json");
+        let res = create_response(&state, StatusCode::CREATED, mime::APPLICATION_JSON, body);
+        Ok((state, res))
+    }
+    .boxed()
 }
 
-fn get_products_handler(state: State) -> Box<HandlerFuture> {
+fn get_products_handler(state: State) -> Pin<Box<HandlerFuture>> {
     use crate::schema::products::dsl::*;
 
     let repo = Repo::borrow_from(&state).clone();
-    let f = repo
-        .run(move |conn| products.load::<Product>(&conn))
-        .then(|result| match result {
+    async move {
+        let result = repo.run(move |conn| products.load::<Product>(&conn)).await;
+        match result {
             Ok(users) => {
                 let body = serde_json::to_string(&users).expect("Failed to serialize users.");
                 let res = create_response(&state, StatusCode::OK, mime::APPLICATION_JSON, body);
-                future::ok((state, res))
+                Ok((state, res))
             }
-            Err(e) => future::err((state, e.into_handler_error())),
-        });
-    Box::new(f)
+            Err(e) => Err((state, e.into_handler_error())),
+        }
+    }
+    .boxed()
 }
 
 fn router(repo: Repo) -> Router {
@@ -104,19 +110,17 @@ where
     e.into_handler_error().with_status(StatusCode::BAD_REQUEST)
 }
 
-fn extract_json<T>(state: &mut State) -> impl Future<Item = T, Error = HandlerError>
+async fn extract_json<T>(state: &mut State) -> Result<T, HandlerError>
 where
     T: serde::de::DeserializeOwned,
 {
-    Body::take_from(state)
-        .concat2()
+    let body = body::to_bytes(Body::take_from(state))
+        .map_err(bad_request)
+        .await?;
+    let b = body.to_vec();
+    from_utf8(&b)
         .map_err(bad_request)
-        .and_then(|body| {
-            let b = body.to_vec();
-            from_utf8(&b)
-                .map_err(bad_request)
-                .and_then(|s| serde_json::from_str::<T>(s).map_err(bad_request))
-        })
+        .and_then(|s| serde_json::from_str::<T>(s).map_err(bad_request))
 }
 
 /// Start a server and use a `Router` to dispatch requests
@@ -136,11 +140,11 @@ fn main() {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use failure::Fail;
     use gotham::test::TestServer;
     use gotham_middleware_diesel::Repo;
     use hyper::StatusCode;
     use std::str;
-    use tokio::runtime;
 
     static DATABASE_URL: &str = ":memory:";
 
@@ -153,7 +157,9 @@ mod tests {
     #[test]
     fn get_empty_products() {
         let repo = Repo::with_test_transactions(DATABASE_URL);
-        runtime::run(repo.run(|conn| embedded_migrations::run(&conn).map_err(|_| ())));
+        let mut runtime = tokio::runtime::Runtime::new().unwrap();
+        let _ = runtime
+            .block_on(repo.run(|conn| embedded_migrations::run(&conn).map_err(|e| e.compat())));
         let test_server = TestServer::new(router(repo)).unwrap();
         let response = test_server
             .client()
@@ -172,7 +178,9 @@ mod tests {
     #[test]
     fn create_and_retrieve_product() {
         let repo = Repo::with_test_transactions(DATABASE_URL);
-        runtime::run(repo.run(|conn| embedded_migrations::run(&conn).map_err(|_| ())));
+        let mut runtime = tokio::runtime::Runtime::new().unwrap();
+        let _ = runtime
+            .block_on(repo.run(|conn| embedded_migrations::run(&conn).map_err(|e| e.compat())));
         let test_server = TestServer::new(router(repo)).unwrap();
 
         //  First we'll insert something into the DB with a post

examples/example_contribution_template/name/Cargo.toml

@@ -11,7 +11,7 @@ publish = false
 [dependencies]
 gotham = { path = "../../../gotham" }
 
-hyper = "0.12"
+hyper = "0.13.1"
 mime = "0.3"
 
 # Add other dependencies if necessary

examples/handlers/async_handlers/Cargo.toml

@@ -9,9 +9,8 @@ publish = false
 gotham = { path = "../../../gotham" }
 gotham_derive = { path = "../../../gotham_derive" }
 
-hyper = "0.12"
+hyper = "0.13.1"
 mime = "0.3"
-futures = "0.1"
+futures = "0.3.1"
 serde = "1"
 serde_derive = "1"
-tokio-core = "0.1"