Repository: seanmonstar/warp (language: rust)

Issue:
Add compression Filters
These would likely need to be `with` filters.

Some hypothetical example:

```rust
let route = warp::fs::dir("static")
    .with(warp::compress::gzip());
```


Discussion / hints:
As an update, quoting from #103:

> I'm hopeful with the changes to support tower (#58), we can just make use of the existing [deflate middleware](https://github.com/carllerche/tower-web/tree/master/src/middleware/deflate).
related to #58 
cc @luciofranco
If we can't integrate with tower's deflate middleware right now, I'd take a stab at implementing this, seems pretty interesting to me!
@ParkMyCar if you're going to look into this, here's a PR that may be of interest to you: https://github.com/seanmonstar/warp/pull/397

I've been thinking about @seanmonstar 's comments on there and tbh I'm not sure a whole lot can be done re: the second point, but I feel the first point is easily solved (and not actually relevant to a compression pr)
@ParkMyCar if you are looking for a deflate crate with AsyncRead/Async Write there's https://docs.rs/async-compression/0.2.0/async_compression/

Fix patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 542277666..e6a53f080 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -35,7 +35,7 @@ jobs:
 
     strategy:
       matrix:
-        build: [stable, beta, nightly, tls, no-default-features]
+        build: [stable, beta, nightly, tls, no-default-features, compression]
 
         include:
           - build: beta
@@ -49,6 +49,8 @@ jobs:
             features: "--features tokio/uds"
           - build: no-default-features
             features: "--no-default-features"
+          - build: compression
+            features: "--features compression"
 
 
     steps:
diff --git a/Cargo.toml b/Cargo.toml
index e9f327e97..731733fcf 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,11 +17,12 @@ edition = "2018"
 features = ["tls"]
 
 [dependencies]
+async-compression = { version = "0.3.1", features = ["brotli", "deflate", "gzip", "stream"], optional = true }
 bytes = "0.5"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 headers = "0.3"
 http = "0.2"
-hyper = "0.13"
+hyper = { version = "0.13", features = ["stream"] }
 log = "0.4"
 mime = "0.3"
 mime_guess = "2.0.0"
@@ -49,6 +50,7 @@ listenfd = "0.3"
 default = ["multipart", "websocket"]
 websocket = ["tokio-tungstenite"]
 tls = ["tokio-rustls"]
+compression = ["async-compression"]
 
 [profile.release]
 codegen-units = 1
@@ -66,6 +68,10 @@ required-features = ["multipart"]
 name = "ws"
 required-features = ["websocket"]
 
+[[example]]
+name = "compression"
+required-features = ["compression"]
+
 [[example]]
 name = "unix_socket"
 required-features = ["tokio/uds"]
diff --git a/README.md b/README.md
index dd19ba1d8..0d4b9196d 100644
--- a/README.md
+++ b/README.md
@@ -20,6 +20,7 @@ Thanks to its `Filter` system, warp provides these out of the box:
 * Static Files and Directories
 * Websockets
 * Access logging
+* Gzip, Deflate, and Brotli compression
 
 Since it builds on top of [hyper](https://hyper.rs), you automatically get:
 
diff --git a/examples/compression.rs b/examples/compression.rs
new file mode 100644
index 000000000..1a52c7a7d
--- /dev/null
+++ b/examples/compression.rs
@@ -0,0 +1,34 @@
+#![deny(warnings)]
+
+use warp::Filter;
+
+#[tokio::main]
+async fn main() {
+    let file = warp::path("todos").and(warp::fs::file("./examples/todos.rs"));
+    // NOTE: You could double compress something by adding a compression
+    // filter here, a la
+    // ```
+    // let file = warp::path("todos")
+    //     .and(warp::fs::file("./examples/todos.rs"))
+    //     .with(warp::compression::brotli());
+    // ```
+    // This would result in a browser error, or downloading a file whose contents
+    // are compressed
+
+    let dir = warp::path("ws_chat").and(warp::fs::file("./examples/websockets_chat.rs"));
+
+    let file_and_dir = warp::get()
+        .and(file.or(dir))
+        .with(warp::compression::gzip());
+
+    let examples = warp::path("ex")
+        .and(warp::fs::dir("./examples/"))
+        .with(warp::compression::deflate());
+
+    // GET /todos => gzip -> toods.rs
+    // GET /ws_chat => gzip -> ws_chat.rs
+    // GET /ex/... => deflate -> ./examples/...
+    let routes = file_and_dir.or(examples);
+
+    warp::serve(routes).run(([127, 0, 0, 1], 3030)).await;
+}
diff --git a/src/filters/compression.rs b/src/filters/compression.rs
new file mode 100644
index 000000000..88c45f86b
--- /dev/null
+++ b/src/filters/compression.rs
@@ -0,0 +1,267 @@
+//! Compression Filters
+//!
+//! Filters that compress the body of a response.
+
+use async_compression::stream::{BrotliEncoder, DeflateEncoder, GzipEncoder};
+use http::header::HeaderValue;
+use hyper::{header::CONTENT_ENCODING, Body};
+
+use crate::filter::{Filter, WrapSealed};
+use crate::reject::IsReject;
+use crate::reply::{Reply, Response};
+
+use self::internal::{CompressionProps, WithCompression};
+
+enum CompressionAlgo {
+    BR,
+    DEFLATE,
+    GZIP,
+}
+
+impl From<CompressionAlgo> for HeaderValue {
+    #[inline]
+    fn from(algo: CompressionAlgo) -> Self {
+        match algo {
+            CompressionAlgo::BR => HeaderValue::from_static("br"),
+            CompressionAlgo::DEFLATE => HeaderValue::from_static("deflate"),
+            CompressionAlgo::GZIP => HeaderValue::from_static("gzip"),
+        }
+    }
+}
+
+/// Compression
+#[derive(Clone, Copy, Debug)]
+pub struct Compression<F> {
+    func: F,
+}
+
+// TODO: The implementation of `gzip()`, `deflate()`, and `brotli()` could be replaced with
+// generics or a macro
+
+/// Create a wrapping filter that compresses the Body of a [`Response`](crate::reply::Response)
+/// using gzip, adding `content-encoding: gzip` to the Response's [`HeaderMap`](hyper::HeaderMap)
+///
+/// # Example
+///
+/// ```
+/// use warp::Filter;
+///
+/// let route = warp::get()
+///     .and(warp::path::end())
+///     .and(warp::fs::file("./README.md"))
+///     .with(warp::compression::gzip());
+/// ```
+pub fn gzip() -> Compression<impl Fn(CompressionProps) -> Response + Copy> {
+    let func = move |mut props: CompressionProps| {
+        let body = Body::wrap_stream(GzipEncoder::new(props.body));
+        props
+            .head
+            .headers
+            .append(CONTENT_ENCODING, CompressionAlgo::GZIP.into());
+        Response::from_parts(props.head, body)
+    };
+    Compression { func }
+}
+
+/// Create a wrapping filter that compresses the Body of a [`Response`](crate::reply::Response)
+/// using deflate, adding `content-encoding: deflate` to the Response's [`HeaderMap`](hyper::HeaderMap)
+///
+/// # Example
+///
+/// ```
+/// use warp::Filter;
+///
+/// let route = warp::get()
+///     .and(warp::path::end())
+///     .and(warp::fs::file("./README.md"))
+///     .with(warp::compression::deflate());
+/// ```
+pub fn deflate() -> Compression<impl Fn(CompressionProps) -> Response + Copy> {
+    let func = move |mut props: CompressionProps| {
+        let body = Body::wrap_stream(DeflateEncoder::new(props.body));
+        props
+            .head
+            .headers
+            .append(CONTENT_ENCODING, CompressionAlgo::DEFLATE.into());
+        Response::from_parts(props.head, body)
+    };
+    Compression { func }
+}
+
+/// Create a wrapping filter that compresses the Body of a [`Response`](crate::reply::Response)
+/// using brotli, adding `content-encoding: br` to the Response's [`HeaderMap`](hyper::HeaderMap)
+///
+/// # Example
+///
+/// ```
+/// use warp::Filter;
+///
+/// let route = warp::get()
+///     .and(warp::path::end())
+///     .and(warp::fs::file("./README.md"))
+///     .with(warp::compression::brotli());
+/// ```
+pub fn brotli() -> Compression<impl Fn(CompressionProps) -> Response + Copy> {
+    let func = move |mut props: CompressionProps| {
+        let body = Body::wrap_stream(BrotliEncoder::new(props.body));
+        props
+            .head
+            .headers
+            .append(CONTENT_ENCODING, CompressionAlgo::BR.into());
+        Response::from_parts(props.head, body)
+    };
+    Compression { func }
+}
+
+impl<FN, F> WrapSealed<F> for Compression<FN>
+where
+    FN: Fn(CompressionProps) -> Response + Clone + Send,
+    F: Filter + Clone + Send,
+    F::Extract: Reply,
+    F::Error: IsReject,
+{
+    type Wrapped = WithCompression<FN, F>;
+
+    fn wrap(&self, filter: F) -> Self::Wrapped {
+        WithCompression {
+            filter,
+            compress: self.clone(),
+        }
+    }
+}
+
+mod internal {
+    use std::future::Future;
+    use std::pin::Pin;
+    use std::task::{Context, Poll};
+
+    use bytes::Bytes;
+    use futures::{ready, Stream, TryFuture};
+    use hyper::Body;
+    use pin_project::pin_project;
+
+    use crate::filter::{Filter, FilterBase, Internal};
+    use crate::reject::IsReject;
+    use crate::reply::{Reply, Response};
+
+    use super::Compression;
+
+    /// A wrapper around any type that implements [`Stream`](futures::Stream) to be
+    /// compatible with async_compression's Stream based encoders
+    #[pin_project]
+    #[derive(Debug)]
+    pub struct CompressableBody<S, E>
+    where
+        E: std::error::Error,
+        S: Stream<Item = Result<Bytes, E>>,
+    {
+        #[pin]
+        body: S,
+    }
+
+    impl<S, E> Stream for CompressableBody<S, E>
+    where
+        E: std::error::Error,
+        S: Stream<Item = Result<Bytes, E>>,
+    {
+        type Item = std::io::Result<Bytes>;
+
+        fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+            use std::io::{Error, ErrorKind};
+
+            let pin = self.project();
+            // TODO: Use `.map_err()` (https://github.com/rust-lang/rust/issues/63514) once it is stabilized
+            S::poll_next(pin.body, cx)
+                .map(|e| e.map(|res| res.map_err(|_| Error::from(ErrorKind::InvalidData))))
+        }
+    }
+
+    impl From<Body> for CompressableBody<Body, hyper::Error> {
+        fn from(body: Body) -> Self {
+            CompressableBody { body }
+        }
+    }
+
+    /// Compression Props
+    #[derive(Debug)]
+    pub struct CompressionProps {
+        pub(super) body: CompressableBody<Body, hyper::Error>,
+        pub(super) head: http::response::Parts,
+    }
+
+    impl From<http::Response<Body>> for CompressionProps {
+        fn from(resp: http::Response<Body>) -> Self {
+            let (head, body) = resp.into_parts();
+            CompressionProps {
+                body: body.into(),
+                head,
+            }
+        }
+    }
+
+    #[allow(missing_debug_implementations)]
+    pub struct Compressed(pub(super) Response);
+
+    impl Reply for Compressed {
+        #[inline]
+        fn into_response(self) -> Response {
+            self.0
+        }
+    }
+
+    #[allow(missing_debug_implementations)]
+    #[derive(Clone, Copy)]
+    pub struct WithCompression<FN, F> {
+        pub(super) compress: Compression<FN>,
+        pub(super) filter: F,
+    }
+
+    impl<FN, F> FilterBase for WithCompression<FN, F>
+    where
+        FN: Fn(CompressionProps) -> Response + Clone + Send,
+        F: Filter + Clone + Send,
+        F::Extract: Reply,
+        F::Error: IsReject,
+    {
+        type Extract = (Compressed,);
+        type Error = F::Error;
+        type Future = WithCompressionFuture<FN, F::Future>;
+
+        fn filter(&self, _: Internal) -> Self::Future {
+            WithCompressionFuture {
+                compress: self.compress.clone(),
+                future: self.filter.filter(Internal),
+            }
+        }
+    }
+
+    #[allow(missing_debug_implementations)]
+    #[pin_project]
+    pub struct WithCompressionFuture<FN, F> {
+        compress: Compression<FN>,
+        #[pin]
+        future: F,
+    }
+
+    impl<FN, F> Future for WithCompressionFuture<FN, F>
+    where
+        FN: Fn(CompressionProps) -> Response,
+        F: TryFuture,
+        F::Ok: Reply,
+        F::Error: IsReject,
+    {
+        type Output = Result<(Compressed,), F::Error>;
+
+        fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+            let pin = self.as_mut().project();
+            let result = ready!(pin.future.try_poll(cx));
+            match result {
+                Ok(reply) => {
+                    let resp = (self.compress.func)(reply.into_response().into());
+                    Poll::Ready(Ok((Compressed(resp),)))
+                }
+                Err(reject) => Poll::Ready(Err(reject)),
+            }
+        }
+    }
+}
diff --git a/src/filters/mod.rs b/src/filters/mod.rs
index 9225601d3..9aee786cf 100644
--- a/src/filters/mod.rs
+++ b/src/filters/mod.rs
@@ -6,6 +6,8 @@
 pub mod addr;
 pub mod any;
 pub mod body;
+#[cfg(feature = "compression")]
+pub mod compression;
 pub mod cookie;
 pub mod cors;
 pub mod ext;
diff --git a/src/lib.rs b/src/lib.rs
index 241ce8443..90ad7d42a 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -106,6 +106,9 @@ pub use self::filter::Filter;
 // This otherwise shows a big dump of re-exports in the doc homepage,
 // with zero context, so just hide it from the docs. Doc examples
 // on each can show that a convenient import exists.
+#[cfg(feature = "compression")]
+#[doc(hidden)]
+pub use self::filters::compression;
 #[cfg(feature = "multipart")]
 #[doc(hidden)]
 pub use self::filters::multipart;


Test patch: