Merge pull request #7 from jtgeibel/new-body

Add support for changes to `conduit::Body`

Author: jtgeibel

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "conduit-hyper"
-version = "0.3.0-alpha.1"
+version = "0.3.0-alpha.2"
 authors = ["Justin Geibel <[email protected]>"]
 license = "MIT OR Apache-2.0"
 description = "Host a conduit based web application on a hyper server"
@@ -9,15 +9,15 @@ readme = "README.md"
 edition = "2018"
 
 [dependencies]
-conduit = "0.9.0-alpha.1"
+conduit = "0.9.0-alpha.2"
 futures = "0.3"
 hyper = "0.13"
 http = "0.2"
 tracing = { version = "0.1", features = ["log"] }
-tokio = { version = "0.2", features = ["blocking", "rt-threaded"] }
+tokio = { version = "0.2", features = ["blocking", "fs", "rt-threaded", "stream"] }
 tower-service = "0.3"
 
 [dev-dependencies]
-conduit-router = "0.9.0-alpha.1"
+conduit-router = "0.9.0-alpha.2"
 env_logger = "0.7"
 tokio = { version = "0.2", features = ["macros"] }

examples/server.rs

@@ -1,6 +1,6 @@
 #![deny(clippy::all)]
 
-use conduit::{header, static_to_body, Handler, RequestExt, Response, ResponseResult};
+use conduit::{header, Body, Handler, RequestExt, Response, ResponseResult};
 use conduit_hyper::Server;
 use conduit_router::RouteBuilder;
 
@@ -35,7 +35,7 @@ fn endpoint(_: &mut dyn RequestExt) -> ResponseResult<http::Error> {
     Response::builder()
         .header(header::CONTENT_TYPE, "text/plain; charset=utf-8")
         .header(header::CONTENT_LENGTH, body.len())
-        .body(static_to_body(body))
+        .body(Body::from_static(body))
 }
 
 fn panic(_: &mut dyn RequestExt) -> ResponseResult<http::Error> {

src/file_stream.rs

@@ -0,0 +1,41 @@
+use std::task::{Context, Poll};
+use std::{io::Error, pin::Pin};
+
+use hyper::body::{Body, Bytes};
+use tokio::{fs::File, io::AsyncRead, stream::Stream};
+
+const BUFFER_SIZE: usize = 8 * 1024;
+
+pub struct FileStream {
+    file: File,
+    buffer: Box<[u8; BUFFER_SIZE]>,
+}
+
+impl FileStream {
+    pub fn from_std(file: std::fs::File) -> Self {
+        let buffer = Box::new([0; BUFFER_SIZE]);
+        let file = File::from_std(file);
+        Self { file, buffer }
+    }
+
+    pub fn into_streamed_body(self) -> Body {
+        Body::wrap_stream(self)
+    }
+}
+
+impl Stream for FileStream {
+    type Item = Result<Bytes, Error>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let Self {
+            ref mut file,
+            ref mut buffer,
+        } = *self;
+        match Pin::new(file).poll_read(cx, &mut buffer[..]) {
+            Poll::Ready(Ok(0)) => Poll::Ready(None),
+            Poll::Ready(Ok(size)) => Poll::Ready(Some(Ok(self.buffer[..size].to_owned().into()))),
+            Poll::Ready(Err(e)) => Poll::Ready(Some(Err(e))),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}

src/lib.rs

@@ -35,17 +35,18 @@
 //! }
 //! #
 //! # use std::{error, io};
-//! # use conduit::{box_error, static_to_body, Response, RequestExt, HandlerResult};
+//! # use conduit::{box_error, Body, Response, RequestExt, HandlerResult};
 //! #
 //! # struct Endpoint();
 //! # impl Handler for Endpoint {
 //! #     fn call(&self, _: &mut dyn RequestExt) -> HandlerResult {
-//! #         Response::builder().body(static_to_body(b"")).map_err(box_error)
+//! #         Response::builder().body(Body::empty()).map_err(box_error)
 //! #     }
 //! # }
 //! ```
 
 mod adaptor;
+mod file_stream;
 mod server;
 mod service;
 #[cfg(test)]
@@ -55,3 +56,18 @@ pub use server::Server;
 pub use service::{BlockingHandler, Service};
 
 type HyperResponse = hyper::Response<hyper::Body>;
+type ConduitResponse = conduit::Response<conduit::Body>;
+use crate::file_stream::FileStream;
+
+/// Turns a `ConduitResponse` into a `HyperResponse`
+fn conduit_into_hyper(response: ConduitResponse) -> HyperResponse {
+    use conduit::Body::*;
+
+    let (parts, body) = response.into_parts();
+    let body = match body {
+        Static(slice) => slice.into(),
+        Owned(vec) => vec.into(),
+        File(file) => FileStream::from_std(file).into_streamed_body(),
+    };
+    HyperResponse::from_parts(parts, body)
+}

src/service.rs

@@ -23,12 +23,12 @@ impl Service {
     /// ```no_run
     /// # use std::sync::Arc;
     /// # use conduit_hyper::{BlockingHandler, Service};
-    /// # use conduit::{box_error, static_to_body, Handler, HandlerResult, RequestExt, Response};
+    /// # use conduit::{box_error, Body, Handler, HandlerResult, RequestExt, Response};
     /// #
     /// # struct Endpoint();
     /// # impl Handler for Endpoint {
     /// #     fn call(&self, _: &mut dyn RequestExt) -> HandlerResult {
-    /// #         Response::builder().body(static_to_body(b"")).map_err(box_error)
+    /// #         Response::builder().body(Body::empty()).map_err(box_error)
     /// #     }
     /// # }
     /// # let app = Endpoint();

src/service/blocking_handler.rs

@@ -1,19 +1,17 @@
 use crate::adaptor::{ConduitRequest, RequestInfo};
 use crate::service::ServiceError;
-use crate::HyperResponse;
+use crate::{conduit_into_hyper, HyperResponse};
 
 use std::net::SocketAddr;
 use std::sync::{
     atomic::{AtomicUsize, Ordering},
     Arc,
 };
 
-use conduit::Handler;
+use conduit::{header, Handler, StatusCode};
 use hyper::{Body, Request, Response};
 use tracing::error;
 
-type ConduitResponse = Response<conduit::Body>;
-
 #[derive(Debug)]
 pub struct BlockingHandler<H: Handler> {
     thread_count: AtomicUsize,
@@ -54,31 +52,20 @@ impl<H: Handler> BlockingHandler<H> {
             let mut request = ConduitRequest::new(&mut request_info, remote_addr);
             handler
                 .call(&mut request)
-                .map(good_response)
+                .map(conduit_into_hyper)
                 .unwrap_or_else(|e| server_error_response(&e.to_string()))
         })
         .await
         .map_err(Into::into)
     }
 }
 
-/// Builds a `hyper::Response` given a `conduit:Response`
-fn good_response(mut response: ConduitResponse) -> HyperResponse {
-    let mut body = Vec::new();
-    if response.body_mut().write_body(&mut body).is_err() {
-        return server_error_response("Error writing body");
-    }
-
-    let (parts, _) = response.into_parts();
-    Response::from_parts(parts, body.into())
-}
-
 /// Logs an error message and returns a generic status 500 response
 fn server_error_response(message: &str) -> HyperResponse {
     error!("Internal Server Error: {}", message);
-    let body = Body::from("Internal Server Error");
+    let body = hyper::Body::from("Internal Server Error");
     Response::builder()
-        .status(500)
+        .status(StatusCode::INTERNAL_SERVER_ERROR)
         .body(body)
         .expect("Unexpected invalid header")
 }
@@ -87,13 +74,13 @@ fn server_error_response(message: &str) -> HyperResponse {
 fn over_capacity_error_response() -> HyperResponse {
     const RETRY_AFTER: u32 = 2;
     error!("Server Capacity Exceeded");
-    let body = Body::from(format!(
+    let body = hyper::Body::from(format!(
         "Service Unavailable: Please retry after {} seconds.",
         RETRY_AFTER
     ));
     Response::builder()
-        .status(503)
-        .header("Retry-After", RETRY_AFTER)
+        .status(StatusCode::SERVICE_UNAVAILABLE)
+        .header(header::RETRY_AFTER, RETRY_AFTER)
         .body(body)
         .expect("Unexpected invalid header")
 }

src/tests.rs

@@ -1,6 +1,4 @@
-use conduit::{
-    box_error, static_to_body, Handler, HandlerResult, RequestExt, Response, StatusCode,
-};
+use conduit::{box_error, Body, Handler, HandlerResult, RequestExt, Response, StatusCode};
 use futures::prelude::*;
 use hyper::{body::to_bytes, service::Service};
 
@@ -12,7 +10,7 @@ impl Handler for OkResult {
     fn call(&self, _req: &mut dyn RequestExt) -> HandlerResult {
         Response::builder()
             .header("ok", "value")
-            .body(static_to_body(b"Hello, world!"))
+            .body(Body::from_static(b"Hello, world!"))
             .map_err(box_error)
     }
 }
@@ -37,7 +35,7 @@ impl Handler for InvalidHeader {
     fn call(&self, _req: &mut dyn RequestExt) -> HandlerResult {
         Response::builder()
             .header("invalid-value", "\r\n")
-            .body(static_to_body(b"discarded"))
+            .body(Body::from_static(b"discarded"))
             .map_err(box_error)
     }
 }
@@ -47,7 +45,7 @@ impl Handler for InvalidStatus {
     fn call(&self, _req: &mut dyn RequestExt) -> HandlerResult {
         Response::builder()
             .status(1000)
-            .body(static_to_body(b""))
+            .body(Body::empty())
             .map_err(box_error)
     }
 }