hyper 1.x support

alttch · Dec 30, 2023 · 06e94fc · 06e94fc
1 parent 328e4cc
commit 06e94fc
Show file tree

Hide file tree

Showing 4 changed files with 90 additions and 39 deletions.
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "hyper-static"
-version = "0.1.6"
+version = "0.2.0"
 edition = "2021"
 rust-version = "1.60"
 license = "MIT"
@@ -10,16 +10,17 @@ description = "A static file handler for Rust/Hyper with minimal logic"
 keywords = ["hyper", "static", "file", "server", "http"]
 
 [dependencies]
-hyper = { version = "0.14.18", features = ["stream"] }
-tokio = { version = "1.15.0", features = ["io-util", "fs"] }
+hyper = { version = "1.1.0" }
+tokio = { version = "1.35.1", features = ["io-util", "fs"] }
 hex = "0.4.3"
-async-stream = "0.3.3"
-chrono = "0.4.19"
-chrono-tz = "0.7.0"
-futures = "0.3.21"
+async-stream = "0.3.5"
+chrono = "0.4.31"
+chrono-tz = "0.8.5"
+futures = "0.3.30"
 bmart-derive = "0.1.2"
-openssl = { version = "0.10.42", optional = true }
-sha2 = { version = "0.10.6", optional = true }
+openssl = { version = "0.10.62", optional = true }
+sha2 = { version = "0.10.8", optional = true }
+http-body-util = "0.1.0"
 
 [features]
 default = ["hashing-openssl"]

diff --git a/README.md b/README.md
@@ -18,44 +18,45 @@ Example:
 use hyper_static::serve::static_file;
 
 // define some hyper handler
-async fn handler(req: Request<Body>) -> Result<Response<Body>, http::Error> {
+async fn handler(req: Request<Incoming>) -> Result<ResponseStreamed, http::Error> {
     // ....
     // serve a file when necessary
     // in a simple way
-    let path = std::path::Path::new("/path/to/file");
+    let mut path = std::path::Path::new("/path/to/files").to_owned();
+    path.push(&req.uri().path()[1..]);
     return match static_file(
         &path,
-        Some("text/html"), // mime type
-        &req.headers(), // hyper request header map
-        65536 // buffer size
+        Some("application/octet-stream"), // mime type
+        &req.headers(),                   // hyper request header map
+        65536,                            // buffer size
     )
     .await
     {
-        Ok(v) => v, // return it
+        Ok(v) => v,         // return it
         Err(e) => e.into(), // transform the error and return
     };
-    // more complicated - analyze errors, e.g. log them
+    //more complicated - analyze errors, e.g. log them
     return match static_file(
         &path,
-        Some("text/html"),
-        &parts.headers,
-        65536
+        Some("application/octet-stream"),
+        &req.headers(),
+        65536,
     )
     .await
     {
         Ok(v) => {
-            debug!(
+            println!(
                 r#""GET {}" {}"#,
-                uri,
+                req.uri(),
                 v.as_ref().map_or(0, |res| res.status().as_u16())
             );
             v
         }
         Err(e) => {
-            let resp: Result<Response<Body>, http::Error> = e.into();
-            warn!(
+            let resp: Result<ResponseStreamed, http::Error> = e.into();
+            eprintln!(
                 r#""GET {}" {}"#,
-                uri,
+                req.uri(),
                 resp.as_ref().map_or(0, |res| res.status().as_u16())
             );
             resp

diff --git a/src/serve.rs b/src/serve.rs
@@ -1,13 +1,18 @@
-use crate::streamer::Streamer;
+use crate::streamer::{Empty, Streamer};
 use bmart_derive::EnumStr;
-use hyper::{http, Body, Response, StatusCode};
+use hyper::{body::Body, http, Response, StatusCode};
+use std::collections::VecDeque;
 use std::io::SeekFrom;
 use std::path::Path;
+use std::pin::Pin;
 use tokio::fs::File;
 use tokio::io::AsyncSeekExt;
 
 pub static DEFAULT_MIME_TYPE: &str = "application/octet-stream";
 
+pub type ResponseStreamed =
+    Response<Pin<Box<dyn Body<Data = VecDeque<u8>, Error = std::io::Error> + 'static + Send>>>;
+
 const TIME_STR: &str = "%a, %d %b %Y %T %Z";
 
 #[derive(Debug, EnumStr, Copy, Clone, Eq, PartialEq)]
@@ -52,15 +57,17 @@ impl Error {
     }
 }
 
-impl From<Error> for Result<Response<Body>, http::Error> {
+impl From<Error> for Result<ResponseStreamed, http::Error> {
     fn from(err: Error) -> Self {
         let code = match err.kind() {
             ErrorKind::Internal => StatusCode::INTERNAL_SERVER_ERROR,
             ErrorKind::Forbidden => StatusCode::FORBIDDEN,
             ErrorKind::NotFound => StatusCode::NOT_FOUND,
             ErrorKind::BadRequest => StatusCode::BAD_REQUEST,
         };
-        Response::builder().status(code).body(Body::empty())
+        Response::builder()
+            .status(code)
+            .body(Box::pin(Empty::new()))
     }
 }
 
@@ -145,7 +152,7 @@ pub async fn static_file<'a>(
     mime_type: Option<&str>,
     headers: &hyper::header::HeaderMap,
     buf_size: usize,
-) -> Result<Result<Response<Body>, http::Error>, Error> {
+) -> Result<Result<ResponseStreamed, http::Error>, Error> {
     macro_rules! forbidden {
         () => {
             return Err(Error::forbidden())
@@ -160,7 +167,7 @@ pub async fn static_file<'a>(
         () => {
             return Ok(Response::builder()
                 .status(StatusCode::NOT_MODIFIED)
-                .body(Body::empty()));
+                .body(Box::pin(Empty::new())));
         };
     }
     let range = if let Some(range_hdr) = headers.get(hyper::header::RANGE) {
@@ -237,19 +244,23 @@ pub async fn static_file<'a>(
                     format!("bytes {}-{}/{}", rn.start, rn.end.unwrap_or(size - 1), size),
                 )
                 .header(hyper::header::CONTENT_LENGTH, part_size)
-                .body(Body::wrap_stream(reader.into_stream_sized(part_size)))
+                .body(Box::pin(http_body_util::StreamBody::new(
+                    reader.into_stream_sized(part_size),
+                )))
         } else {
             Response::builder()
                 .status(StatusCode::RANGE_NOT_SATISFIABLE)
                 .header(hyper::header::ACCEPT_RANGES, "bytes")
                 .header(hyper::header::CONTENT_RANGE, format!("*/{}", size))
-                .body(Body::empty())
+                .body(Box::pin(Empty::new()))
         }
     } else {
         let reader = Streamer::new(f, buf_size);
         resp!(StatusCode::OK, last_modified, etag, mime_type)
             .header(hyper::header::CONTENT_LENGTH, size)
-            .body(Body::wrap_stream(reader.into_stream()))
+            .body(Box::pin(http_body_util::StreamBody::new(
+                reader.into_stream(),
+            )))
     })
 }
 

diff --git a/src/streamer.rs b/src/streamer.rs
@@ -1,9 +1,48 @@
 use async_stream::stream;
 use futures::{Stream, StreamExt};
+use hyper::body::{Body, Buf, Frame, SizeHint};
+use std::collections::VecDeque;
 use std::io::Error;
+use std::marker::PhantomData;
 use std::pin::Pin;
+use std::task::{Context, Poll};
 use tokio::io::{AsyncRead, AsyncReadExt};
 
+#[derive(Default)]
+pub struct Empty<D> {
+    _marker: PhantomData<fn() -> D>,
+}
+
+impl<D> Empty<D>
+where
+    D: Default,
+{
+    pub fn new() -> Self {
+        Self::default()
+    }
+}
+
+impl<D: Buf> Body for Empty<D> {
+    type Data = D;
+    type Error = Error;
+
+    #[inline]
+    fn poll_frame(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
+        Poll::Ready(None)
+    }
+
+    fn is_end_stream(&self) -> bool {
+        true
+    }
+
+    fn size_hint(&self) -> SizeHint {
+        SizeHint::with_exact(0)
+    }
+}
+
 pub struct Streamer<R>
 where
     R: AsyncRead + Unpin + Send + 'static,
@@ -22,7 +61,7 @@ where
     }
     pub fn into_stream(
         mut self,
-    ) -> Pin<Box<impl ?Sized + Stream<Item = Result<Vec<u8>, Error>> + 'static>> {
+    ) -> Pin<Box<impl ?Sized + Stream<Item = Result<Frame<VecDeque<u8>>, Error>> + 'static>> {
         let stream = stream! {
             loop {
                 let mut buf = vec![0; self.buf_size];
@@ -31,7 +70,7 @@ where
                     break
                 }
                 buf.truncate(r);
-                yield Ok(buf);
+                yield Ok(Frame::data(buf.into()));
             }
         };
         stream.boxed()
@@ -41,7 +80,7 @@ where
     pub fn into_stream_sized(
         mut self,
         max_length: u64,
-    ) -> Pin<Box<impl ?Sized + Stream<Item = Result<Vec<u8>, Error>> + 'static>> {
+    ) -> Pin<Box<impl ?Sized + Stream<Item = Result<Frame<VecDeque<u8>>, Error>> + 'static>> {
         let stream = stream! {
             let mut remaining = max_length;
             loop {
@@ -57,10 +96,9 @@ where
                 let r = self.reader.read(&mut buf).await?;
                 if r == 0 {
                     break;
-                } else {
-                    buf.truncate(r);
-                    yield Ok(buf);
                 }
+                buf.truncate(r);
+                yield Ok(Frame::data(buf.into()));
                 remaining -= r as u64;
             }
         };