Upgrade to hyper-0.13

Author: jtgeibel

Cargo.toml

@@ -10,15 +10,15 @@ edition = "2018"
 
 [dependencies]
 conduit = "0.8"
-futures-preview = "0.3.0-alpha.19"
-hyper = { version = "0.13.0-alpha.4", features = ["unstable-stream"] }
-http = "0.1"
+futures = "0.3"
+hyper = "0.13"
+http = "0.2"
 tracing = { version = "0.1", features = ["log"] }
 semver = "0.5" # Must match version in conduit for now
-tokio-executor = { version = "0.2.0-alpha.6", features = ["threadpool"] }
-tower-service = "0.3.0-alpha.2"
+tokio = { version = "0.2", features = ["blocking", "rt-threaded"] }
+tower-service = "0.3"
 
 [dev-dependencies]
 conduit-router = "0.8"
-env_logger = "0.6"
-tokio = { version = "0.2.0-alpha.2", default-features = false, features = ["rt-full"] }
+env_logger = "0.7"
+tokio = { version = "0.2", features = ["macros"] }

examples/server.rs

@@ -3,28 +3,20 @@
 use conduit::{Handler, Request, Response};
 use conduit_hyper::Server;
 use conduit_router::RouteBuilder;
-use tokio::runtime;
 
 use std::collections::HashMap;
 use std::io::{Cursor, Error};
 use std::thread::sleep;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     env_logger::init();
 
     let app = build_conduit_handler();
     let addr = ([127, 0, 0, 1], 12345).into();
-    let server = Server::bind(&addr, app);
-
-    let rt = runtime::Builder::new()
-        // Set the max number of concurrent requests (tokio defaults to 100)
-        .blocking_threads(2)
-        .build()
-        .unwrap();
-    rt.spawn(async {
-        server.await;
-    });
-    rt.shutdown_on_idle();
+
+    // FIXME: Set limit on number of blocking tasks
+    Server::serve(&addr, app).await;
 }
 
 fn build_conduit_handler() -> impl Handler {

src/adaptor.rs

@@ -15,18 +15,18 @@ use std::path::{Component, Path, PathBuf};
 
 use conduit::{Extensions, Headers, Host, Method, Request, Scheme};
 use http::{request::Parts as HttpParts, HeaderMap};
-use hyper::{Chunk, Method as HyperMethod, Version as HttpVersion};
+use hyper::{body::Bytes, Method as HyperMethod, Version as HttpVersion};
 use semver::Version;
 
 /// Owned data consumed by the background thread
 ///
 /// `ConduitRequest` cannot be sent between threads, so the needed request data
 /// is extracted from hyper on a core thread and taken by the background thread.
-pub(crate) struct RequestInfo(Option<(Parts, Chunk)>);
+pub(crate) struct RequestInfo(Option<(Parts, Bytes)>);
 
 impl RequestInfo {
     /// Save the request info that can be sent between threads
-    pub(crate) fn new(parts: HttpParts, body: Chunk) -> Self {
+    pub(crate) fn new(parts: HttpParts, body: Bytes) -> Self {
         let tuple = (Parts(parts), body);
         Self(Some(tuple))
     }
@@ -38,7 +38,7 @@ impl RequestInfo {
     /// # Panics
     ///
     /// Panics if called more than once on a value.
-    fn take(&mut self) -> (Parts, Chunk) {
+    fn take(&mut self) -> (Parts, Bytes) {
         self.0.take().expect("called take multiple times")
     }
 }
@@ -94,7 +94,7 @@ pub(crate) struct ConduitRequest {
     parts: Parts,
     path: String,
     remote_addr: SocketAddr,
-    body: Cursor<Chunk>,
+    body: Cursor<Bytes>,
     extensions: Extensions, // makes struct non-Send
 }
 
@@ -148,6 +148,8 @@ impl Request for ConduitRequest {
             HttpVersion::HTTP_10 => version(1, 0),
             HttpVersion::HTTP_11 => version(1, 1),
             HttpVersion::HTTP_2 => version(2, 0),
+            HttpVersion::HTTP_3 => version(3, 0),
+            _ => version(0, 0),
         }
     }
 

src/lib.rs

@@ -1,4 +1,4 @@
-#![deny(warnings, clippy::all, missing_debug_implementations)]
+#![deny(warnings, clippy::all, missing_debug_implementations, rust_2018_idioms)]
 #![allow(clippy::needless_doctest_main, clippy::unknown_clippy_lints)] // using the tokio::main attribute
 
 //! A wrapper for integrating `hyper 0.13` with a `conduit 0.8` blocking application stack.
@@ -16,14 +16,13 @@
 //! ```no_run
 //! use conduit::Handler;
 //! use conduit_hyper::Server;
-//! use futures::executor::block_on;
 //! use tokio::runtime::Runtime;
 //!
 //! #[tokio::main]
 //! async fn main() {
 //!     let app = build_conduit_handler();
 //!     let addr = ([127, 0, 0, 1], 12345).into();
-//!     let server = Server::bind(&addr, app);
+//!     let server = Server::serve(&addr, app);
 //!
 //!     server.await;
 //! }

src/server.rs

@@ -19,7 +19,7 @@ impl Server {
     /// `tokio::Runtime` it is not possible to furter configure the `hyper::Server`.  If more
     /// control, such as configuring a graceful shutdown is necessary, then call
     /// `Service::from_conduit` instead.
-    pub fn bind<H: conduit::Handler>(addr: &SocketAddr, handler: H) -> impl Future {
+    pub fn serve<H: conduit::Handler>(addr: &SocketAddr, handler: H) -> impl Future {
         let handler = Arc::new(handler);
         let make_service = make_service_fn(move |socket: &AddrStream| {
             let handler = handler.clone();

src/service.rs

@@ -4,10 +4,12 @@ use std::future::Future;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use futures::prelude::*;
-use hyper::{service, Body, Error, Request, Response, StatusCode};
+use hyper::{service, Body, Request, Response, StatusCode};
 use tracing::error;
 
+mod error;
+pub use error::ServiceError;
+
 /// A builder for a `hyper::Service`.
 #[derive(Debug)]
 pub struct Service;
@@ -54,10 +56,10 @@ impl Service {
         impl tower_service::Service<
             Request<Body>,
             Response = Response<Body>,
-            Error = Error,
-            Future = impl Future<Output = Result<Response<Body>, Error>> + Send + 'static,
+            Error = ServiceError,
+            Future = impl Future<Output = Result<Response<Body>, ServiceError>> + Send + 'static,
         >,
-        Error,
+        ServiceError,
     > {
         Ok(service::service_fn(move |request: Request<Body>| {
             blocking_handler(handler.clone(), request, remote_addr)
@@ -70,26 +72,22 @@ pub(crate) async fn blocking_handler<H: conduit::Handler>(
     handler: Arc<H>,
     request: Request<Body>,
     remote_addr: SocketAddr,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<Body>, ServiceError> {
     let (parts, body) = request.into_parts();
-    let full_body = body.try_concat().await?;
-    let mut request_info = RequestInfo::new(parts, full_body);
 
-    let future = future::poll_fn(move |_| {
-        tokio_executor::threadpool::blocking(|| {
-            let mut request = ConduitRequest::new(&mut request_info, remote_addr);
-            handler
-                .call(&mut request)
-                .map(good_response)
-                .unwrap_or_else(|e| error_response(&e.to_string()))
-        })
-        .map_err(|_| panic!("The threadpool shut down"))
-    });
+    let full_body = hyper::body::to_bytes(body).await?;
+    let mut request_info = RequestInfo::new(parts, full_body);
 
-    // Spawn as a new top-level task, otherwise the parent task is blocked as well
-    let (future, handle) = future.remote_handle();
-    tokio_executor::spawn(future);
-    handle.await
+    // FIXME: Provide a configurable limit on the number of blocking tasks
+    tokio::task::spawn_blocking(move || {
+        let mut request = ConduitRequest::new(&mut request_info, remote_addr);
+        handler
+            .call(&mut request)
+            .map(good_response)
+            .unwrap_or_else(|e| error_response(&e.to_string()))
+    })
+    .await
+    .map_err(Into::into)
 }
 
 /// Builds a `hyper::Response` given a `conduit:Response`
@@ -104,15 +102,15 @@ fn good_response(mut response: conduit::Response) -> Response<Body> {
         Ok(s) => s,
         Err(e) => return error_response(&e.to_string()),
     };
-    builder.status(status);
 
     for (key, values) in response.headers {
         for value in values {
-            builder.header(key.as_str(), value.as_str());
+            builder = builder.header(key.as_str(), value.as_str());
         }
     }
 
     builder
+        .status(status)
         .body(body.into())
         .unwrap_or_else(|e| error_response(&e.to_string()))
 }

src/service/error.rs

@@ -0,0 +1,32 @@
+use std::error::Error;
+
+use tokio::task::JoinError;
+
+#[derive(Debug)]
+pub enum ServiceError {
+    JoinError(std::io::Error),
+    Hyper(hyper::Error),
+}
+
+impl Error for ServiceError {}
+
+impl std::fmt::Display for ServiceError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ServiceError::JoinError(e) => e.fmt(f),
+            ServiceError::Hyper(e) => e.fmt(f),
+        }
+    }
+}
+
+impl From<JoinError> for ServiceError {
+    fn from(e: JoinError) -> Self {
+        ServiceError::JoinError(e.into())
+    }
+}
+
+impl From<hyper::Error> for ServiceError {
+    fn from(e: hyper::Error) -> Self {
+        ServiceError::Hyper(e)
+    }
+}