Merge remote-tracking branch 'origin/master' into to-forgive-divine

Cargo.toml

@@ -101,6 +101,7 @@ brotli2 = "0.3.2"
 criterion = "0.3"
 env_logger = "0.8"
 flate2 = "1.0.13"
+zstd = "0.7"
 rand = "0.8"
 rcgen = "0.8"
 serde_derive = "1.0"

actix-files/CHANGES.md

@@ -3,9 +3,11 @@
 ## Unreleased - 2021-xx-xx
 * `NamedFile` now implements `ServiceFactory` and `HttpServiceFactory` making it much more useful in routing. For example, it can be used directly as a default service. [#2135]
 * For symbolic links, `Content-Disposition` header no longer shows the filename of the original file. [#2156]
+* `Files::redirect_to_slash_directory()` now works as expected when used with `Files::show_files_listing()`. [#2225]
 
 [#2135]: https://github.com/actix/actix-web/pull/2135
 [#2156]: https://github.com/actix/actix-web/pull/2156
+[#2225]: https://github.com/actix/actix-web/pull/2225
 
 
 ## 0.6.0-beta.4 - 2021-04-02

actix-files/src/lib.rs

@@ -632,7 +632,7 @@ mod tests {
 
     #[actix_rt::test]
     async fn test_redirect_to_slash_directory() {
-        // should not redirect if no index
+        // should not redirect if no index and files listing is disabled
         let srv = test::init_service(
             App::new().service(Files::new("/", ".").redirect_to_slash_directory()),
         )
@@ -654,6 +654,19 @@ mod tests {
         let resp = test::call_service(&srv, req).await;
         assert_eq!(resp.status(), StatusCode::FOUND);
 
+        // should redirect if files listing is enabled
+        let srv = test::init_service(
+            App::new().service(
+                Files::new("/", ".")
+                    .show_files_listing()
+                    .redirect_to_slash_directory(),
+            ),
+        )
+        .await;
+        let req = TestRequest::with_uri("/tests").to_request();
+        let resp = test::call_service(&srv, req).await;
+        assert_eq!(resp.status(), StatusCode::FOUND);
+
         // should not redirect if the path is wrong
         let req = TestRequest::with_uri("/not_existing").to_request();
         let resp = test::call_service(&srv, req).await;

actix-files/src/service.rs

@@ -89,8 +89,10 @@ impl Service<ServiceRequest> for FilesService {
         }
 
         if path.is_dir() {
-            if let Some(ref redir_index) = self.index {
+            if self.redirect_to_slash
-                if self.redirect_to_slash && !req.path().ends_with('/') {
+                && !req.path().ends_with('/')
+                && (self.index.is_some() || self.show_index)
+            {
                 let redirect_to = format!("{}/", req.path());
 
                 return Box::pin(ok(req.into_response(
@@ -100,6 +102,7 @@ impl Service<ServiceRequest> for FilesService {
                 )));
             }
 
+            if let Some(ref redir_index) = self.index {
                 let path = path.join(redir_index);
 
                 match NamedFile::open(path) {

actix-http/CHANGES.md

@@ -9,6 +9,7 @@
 * Re-export `ContentEncoding` and `ConnectionType` at the crate root. [#2171]
 * `Response::into_body` that consumes response and returns body type. [#2201]
 * `impl Default` for `Response`. [#2201]
+* Add zstd support for `ContentEncoding`. [#2244]
 
 ### Changed
 * The `MessageBody` trait now has an associated `Error` type. [#2183]
@@ -39,6 +40,8 @@
 [#2205]: https://github.com/actix/actix-web/pull/2205
 [#2215]: https://github.com/actix/actix-web/pull/2215
 [#2224]: https://github.com/actix/actix-web/pull/2224
+[#2244]: https://github.com/actix/actix-web/pull/2244
+
 
 
 ## 3.0.0-beta.6 - 2021-04-17

actix-http/Cargo.toml

@@ -32,7 +32,7 @@ openssl = ["actix-tls/openssl"]
 rustls = ["actix-tls/rustls"]
 
 # enable compression support
-compress = ["flate2", "brotli2"]
+compress = ["flate2", "brotli2", "zstd"]
 
 # trust-dns as client dns resolver
 trust-dns = ["trust-dns-resolver"]
@@ -76,6 +76,7 @@ tokio = { version = "1.2", features = ["sync"] }
 # compression
 brotli2 = { version="0.3.2", optional = true }
 flate2 = { version = "1.0.13", optional = true }
+zstd = { version = "0.7", optional = true }
 
 trust-dns-resolver = { version = "0.20.0", optional = true }
 

actix-http/src/encoding/decoder.rs

@@ -12,6 +12,7 @@ use brotli2::write::BrotliDecoder;
 use bytes::Bytes;
 use flate2::write::{GzDecoder, ZlibDecoder};
 use futures_core::{ready, Stream};
+use zstd::stream::write::Decoder as ZstdDecoder;
 
 use crate::{
     encoding::Writer,
@@ -45,6 +46,12 @@ where
             ContentEncoding::Gzip => Some(ContentDecoder::Gzip(Box::new(
                 GzDecoder::new(Writer::new()),
             ))),
+            ContentEncoding::Zstd => Some(ContentDecoder::Zstd(Box::new(
+                ZstdDecoder::new(Writer::new()).expect(
+                    "Failed to create zstd decoder. This is a bug. \
+                         Please report it to the actix-web repository.",
+                ),
+            ))),
             _ => None,
         };
 
@@ -144,6 +151,9 @@ enum ContentDecoder {
     Deflate(Box<ZlibDecoder<Writer>>),
     Gzip(Box<GzDecoder<Writer>>),
     Br(Box<BrotliDecoder<Writer>>),
+    // We need explicit 'static lifetime here because ZstdDecoder need lifetime
+    // argument, and we use `spawn_blocking` in `Decoder::poll_next` that require `FnOnce() -> R + Send + 'static`
+    Zstd(Box<ZstdDecoder<'static, Writer>>),
 }
 
 impl ContentDecoder {
@@ -186,6 +196,18 @@ impl ContentDecoder {
                 }
                 Err(e) => Err(e),
             },
+
+            ContentDecoder::Zstd(ref mut decoder) => match decoder.flush() {
+                Ok(_) => {
+                    let b = decoder.get_mut().take();
+                    if !b.is_empty() {
+                        Ok(Some(b))
+                    } else {
+                        Ok(None)
+                    }
+                }
+                Err(e) => Err(e),
+            },
         }
     }
 
@@ -232,6 +254,20 @@ impl ContentDecoder {
                 }
                 Err(e) => Err(e),
             },
+
+            ContentDecoder::Zstd(ref mut decoder) => match decoder.write_all(&data) {
+                Ok(_) => {
+                    decoder.flush()?;
+
+                    let b = decoder.get_mut().take();
+                    if !b.is_empty() {
+                        Ok(Some(b))
+                    } else {
+                        Ok(None)
+                    }
+                }
+                Err(e) => Err(e),
+            },
         }
     }
 }

actix-http/src/encoding/encoder.rs

@@ -15,6 +15,7 @@ use derive_more::Display;
 use flate2::write::{GzEncoder, ZlibEncoder};
 use futures_core::ready;
 use pin_project::pin_project;
+use zstd::stream::write::Encoder as ZstdEncoder;
 
 use crate::{
     body::{Body, BodySize, BoxAnyBody, MessageBody, ResponseBody},
@@ -237,6 +238,9 @@ enum ContentEncoder {
     Deflate(ZlibEncoder<Writer>),
     Gzip(GzEncoder<Writer>),
     Br(BrotliEncoder<Writer>),
+    // We need explicit 'static lifetime here because ZstdEncoder need lifetime
+    // argument, and we use `spawn_blocking` in `Encoder::poll_next` that require `FnOnce() -> R + Send + 'static`
+    Zstd(ZstdEncoder<'static, Writer>),
 }
 
 impl ContentEncoder {
@@ -253,6 +257,10 @@ impl ContentEncoder {
             ContentEncoding::Br => {
                 Some(ContentEncoder::Br(BrotliEncoder::new(Writer::new(), 3)))
             }
+            ContentEncoding::Zstd => {
+                let encoder = ZstdEncoder::new(Writer::new(), 3).ok()?;
+                Some(ContentEncoder::Zstd(encoder))
+            }
             _ => None,
         }
     }
@@ -263,6 +271,7 @@ impl ContentEncoder {
             ContentEncoder::Br(ref mut encoder) => encoder.get_mut().take(),
             ContentEncoder::Deflate(ref mut encoder) => encoder.get_mut().take(),
             ContentEncoder::Gzip(ref mut encoder) => encoder.get_mut().take(),
+            ContentEncoder::Zstd(ref mut encoder) => encoder.get_mut().take(),
         }
     }
 
@@ -280,6 +289,10 @@ impl ContentEncoder {
                 Ok(writer) => Ok(writer.buf.freeze()),
                 Err(err) => Err(err),
             },
+            ContentEncoder::Zstd(encoder) => match encoder.finish() {
+                Ok(writer) => Ok(writer.buf.freeze()),
+                Err(err) => Err(err),
+            },
         }
     }
 
@@ -306,6 +319,13 @@ impl ContentEncoder {
                     Err(err)
                 }
             },
+            ContentEncoder::Zstd(ref mut encoder) => match encoder.write_all(data) {
+                Ok(_) => Ok(()),
+                Err(err) => {
+                    trace!("Error decoding ztsd encoding: {}", err);
+                    Err(err)
+                }
+            },
         }
     }
 }

actix-http/src/h2/dispatcher.rs

@@ -11,49 +11,36 @@ use std::{
 
 use actix_codec::{AsyncRead, AsyncWrite};
 use actix_service::Service;
+use actix_utils::future::poll_fn;
 use bytes::{Bytes, BytesMut};
 use futures_core::ready;
-use h2::{
+use h2::server::{Connection, SendResponse};
-    server::{Connection, SendResponse},
-    SendStream,
-};
 use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, DATE, TRANSFER_ENCODING};
 use log::{error, trace};
+use pin_project_lite::pin_project;
 
 use crate::{
-    body::{AnyBody, Body, BodySize, MessageBody},
+    body::{AnyBody, BodySize, MessageBody},
     config::ServiceConfig,
-    error::DispatchError,
     service::HttpFlow,
     OnConnectData, Payload, Request, Response, ResponseHead,
 };
 
 const CHUNK_SIZE: usize = 16_384;
 
-/// Dispatcher for HTTP/2 protocol.
+pin_project! {
-#[pin_project::pin_project]
+    /// Dispatcher for HTTP/2 protocol.
-pub struct Dispatcher<T, S, B, X, U>
+    pub struct Dispatcher<T, S, B, X, U> {
-where
-    T: AsyncRead + AsyncWrite + Unpin,
-    S: Service<Request>,
-    B: MessageBody,
-{
         flow: Rc<HttpFlow<S, X, U>>,
         connection: Connection<T, Bytes>,
         on_connect_data: OnConnectData,
         config: ServiceConfig,
         peer_addr: Option<net::SocketAddr>,
         _phantom: PhantomData<B>,
+    }
 }
 
-impl<T, S, B, X, U> Dispatcher<T, S, B, X, U>
+impl<T, S, B, X, U> Dispatcher<T, S, B, X, U> {
-where
-    T: AsyncRead + AsyncWrite + Unpin,
-    S: Service<Request>,
-    S::Error: Into<Response<AnyBody>>,
-    S::Response: Into<Response<B>>,
-    B: MessageBody,
-{
     pub(crate) fn new(
         flow: Rc<HttpFlow<S, X, U>>,
         connection: Connection<T, Bytes>,
@@ -61,7 +48,7 @@ where
         config: ServiceConfig,
         peer_addr: Option<net::SocketAddr>,
     ) -> Self {
-        Dispatcher {
+        Self {
             flow,
             config,
             peer_addr,
@@ -77,26 +64,22 @@ where
     T: AsyncRead + AsyncWrite + Unpin,
 
     S: Service<Request>,
-    S::Error: Into<Response<AnyBody>> + 'static,
+    S::Error: Into<Response<AnyBody>>,
     S::Future: 'static,
-    S::Response: Into<Response<B>> + 'static,
+    S::Response: Into<Response<B>>,
 
-    B: MessageBody + 'static,
+    B: MessageBody,
     B::Error: Into<Box<dyn StdError>>,
 {
-    type Output = Result<(), DispatchError>;
+    type Output = Result<(), crate::error::DispatchError>;
 
     #[inline]
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.get_mut();
 
-        loop {
+        while let Some((req, tx)) =
-            match ready!(Pin::new(&mut this.connection).poll_accept(cx)) {
+            ready!(Pin::new(&mut this.connection).poll_accept(cx)?)
-                None => return Poll::Ready(Ok(())),
+        {
-
-                Some(Err(err)) => return Poll::Ready(Err(err.into())),
-
-                Some(Ok((req, res))) => {
             let (parts, body) = req.into_parts();
             let pl = crate::h2::Payload::new(body);
             let pl = Payload::<crate::payload::PayloadStream>::H2(pl);
@@ -112,53 +95,116 @@ where
             // merge on_connect_ext data into request extensions
             this.on_connect_data.merge_into(&mut req);
 
-                    let svc = ServiceResponse {
+            let fut = this.flow.service.call(req);
-                        state: ServiceResponseState::ServiceCall(
+            let config = this.config.clone();
-                            this.flow.service.call(req),
+
-                            Some(res),
+            // multiplex request handling with spawn task
-                        ),
+            actix_rt::spawn(async move {
-                        config: this.config.clone(),
+                // resolve service call and send response.
-                        buffer: None,
+                let res = match fut.await {
-                        _phantom: PhantomData,
+                    Ok(res) => handle_response(res.into(), tx, config).await,
+                    Err(err) => {
+                        let res: Response<AnyBody> = err.into();
+                        handle_response(res, tx, config).await
+                    }
                 };
 
-                    actix_rt::spawn(svc);
+                // log error.
+                if let Err(err) = res {
+                    match err {
+                        DispatchError::SendResponse(err) => {
+                            trace!("Error sending HTTP/2 response: {:?}", err)
+                        }
+                        DispatchError::SendData(err) => warn!("{:?}", err),
+                        DispatchError::ResponseBody(err) => {
+                            error!("Response payload stream error: {:?}", err)
                         }
                     }
                 }
+            });
+        }
+
+        Poll::Ready(Ok(()))
     }
 }
 
-#[pin_project::pin_project]
+enum DispatchError {
-struct ServiceResponse<F, I, E, B> {
+    SendResponse(h2::Error),
-    #[pin]
+    SendData(h2::Error),
-    state: ServiceResponseState<F, B>,
+    ResponseBody(Box<dyn StdError>),
+}
+
+async fn handle_response<B>(
+    res: Response<B>,
+    mut tx: SendResponse<Bytes>,
     config: ServiceConfig,
-    buffer: Option<Bytes>,
+) -> Result<(), DispatchError>
-    _phantom: PhantomData<(I, E)>,
-}
-
-#[pin_project::pin_project(project = ServiceResponseStateProj)]
-enum ServiceResponseState<F, B> {
-    ServiceCall(#[pin] F, Option<SendResponse<Bytes>>),
-    SendPayload(SendStream<Bytes>, #[pin] B),
-    SendErrorPayload(SendStream<Bytes>, #[pin] Body),
-}
-
-impl<F, I, E, B> ServiceResponse<F, I, E, B>
 where
-    F: Future<Output = Result<I, E>>,
-    E: Into<Response<AnyBody>>,
-    I: Into<Response<B>>,
-
     B: MessageBody,
     B::Error: Into<Box<dyn StdError>>,
 {
-    fn prepare_response(
+    let (res, body) = res.replace_body(());
-        &self,
+
+    // prepare response.
+    let mut size = body.size();
+    let res = prepare_response(config, res.head(), &mut size);
+    let eof = size.is_eof();
+
+    // send response head and return on eof.
+    let mut stream = tx
+        .send_response(res, eof)
+        .map_err(DispatchError::SendResponse)?;
+
+    if eof {
+        return Ok(());
+    }
+
+    // poll response body and send chunks to client.
+    actix_rt::pin!(body);
+
+    while let Some(res) = poll_fn(|cx| body.as_mut().poll_next(cx)).await {
+        let mut chunk = res.map_err(|err| DispatchError::ResponseBody(err.into()))?;
+
+        'send: loop {
+            // reserve enough space and wait for stream ready.
+            stream.reserve_capacity(cmp::min(chunk.len(), CHUNK_SIZE));
+
+            match poll_fn(|cx| stream.poll_capacity(cx)).await {
+                // No capacity left. drop body and return.
+                None => return Ok(()),
+                Some(res) => {
+                    // Split chuck to writeable size and send to client.
+                    let cap = res.map_err(DispatchError::SendData)?;
+
+                    let len = chunk.len();
+                    let bytes = chunk.split_to(cmp::min(cap, len));
+
+                    stream
+                        .send_data(bytes, false)
+                        .map_err(DispatchError::SendData)?;
+
+                    // Current chuck completely sent. break send loop and poll next one.
+                    if chunk.is_empty() {
+                        break 'send;
+                    }
+                }
+            }
+        }
+    }
+
+    // response body streaming finished. send end of stream and return.
+    stream
+        .send_data(Bytes::new(), true)
+        .map_err(DispatchError::SendData)?;
+
+    Ok(())
+}
+
+fn prepare_response(
+    config: ServiceConfig,
     head: &ResponseHead,
     size: &mut BodySize,
-    ) -> http::Response<()> {
+) -> http::Response<()> {
     let mut has_date = false;
     let mut skip_len = size != &BodySize::Stream;
 
@@ -211,7 +257,7 @@ where
     // set date header
     if !has_date {
         let mut bytes = BytesMut::with_capacity(29);
-            self.config.set_date_header(&mut bytes);
+        config.set_date_header(&mut bytes);
         res.headers_mut().insert(
             DATE,
             // SAFETY: serialized date-times are known ASCII strings
@@ -220,188 +266,4 @@ where
     }
 
     res
-    }
-}
-
-impl<F, I, E, B> Future for ServiceResponse<F, I, E, B>
-where
-    F: Future<Output = Result<I, E>>,
-    E: Into<Response<AnyBody>>,
-    I: Into<Response<B>>,
-
-    B: MessageBody,
-    B::Error: Into<Box<dyn StdError>>,
-{
-    type Output = ();
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut this = self.as_mut().project();
-
-        match this.state.project() {
-            ServiceResponseStateProj::ServiceCall(call, send) => {
-                match ready!(call.poll(cx)) {
-                    Ok(res) => {
-                        let (res, body) = res.into().replace_body(());
-
-                        let mut send = send.take().unwrap();
-                        let mut size = body.size();
-                        let h2_res =
-                            self.as_mut().prepare_response(res.head(), &mut size);
-                        this = self.as_mut().project();
-
-                        let stream = match send.send_response(h2_res, size.is_eof()) {
-                            Err(e) => {
-                                trace!("Error sending HTTP/2 response: {:?}", e);
-                                return Poll::Ready(());
-                            }
-                            Ok(stream) => stream,
-                        };
-
-                        if size.is_eof() {
-                            Poll::Ready(())
-                        } else {
-                            this.state
-                                .set(ServiceResponseState::SendPayload(stream, body));
-                            self.poll(cx)
-                        }
-                    }
-
-                    Err(err) => {
-                        let res: Response<AnyBody> = err.into();
-                        let (res, body) = res.replace_body(());
-
-                        let mut send = send.take().unwrap();
-                        let mut size = body.size();
-                        let h2_res =
-                            self.as_mut().prepare_response(res.head(), &mut size);
-                        this = self.as_mut().project();
-
-                        let stream = match send.send_response(h2_res, size.is_eof()) {
-                            Err(e) => {
-                                trace!("Error sending HTTP/2 response: {:?}", e);
-                                return Poll::Ready(());
-                            }
-                            Ok(stream) => stream,
-                        };
-
-                        if size.is_eof() {
-                            Poll::Ready(())
-                        } else {
-                            this.state.set(ServiceResponseState::SendErrorPayload(
-                                stream, body,
-                            ));
-                            self.poll(cx)
-                        }
-                    }
-                }
-            }
-
-            ServiceResponseStateProj::SendPayload(ref mut stream, ref mut body) => {
-                loop {
-                    match this.buffer {
-                        Some(ref mut buffer) => match ready!(stream.poll_capacity(cx)) {
-                            None => return Poll::Ready(()),
-
-                            Some(Ok(cap)) => {
-                                let len = buffer.len();
-                                let bytes = buffer.split_to(cmp::min(cap, len));
-
-                                if let Err(e) = stream.send_data(bytes, false) {
-                                    warn!("{:?}", e);
-                                    return Poll::Ready(());
-                                } else if !buffer.is_empty() {
-                                    let cap = cmp::min(buffer.len(), CHUNK_SIZE);
-                                    stream.reserve_capacity(cap);
-                                } else {
-                                    this.buffer.take();
-                                }
-                            }
-
-                            Some(Err(e)) => {
-                                warn!("{:?}", e);
-                                return Poll::Ready(());
-                            }
-                        },
-
-                        None => match ready!(body.as_mut().poll_next(cx)) {
-                            None => {
-                                if let Err(e) = stream.send_data(Bytes::new(), true) {
-                                    warn!("{:?}", e);
-                                }
-                                return Poll::Ready(());
-                            }
-
-                            Some(Ok(chunk)) => {
-                                stream
-                                    .reserve_capacity(cmp::min(chunk.len(), CHUNK_SIZE));
-                                *this.buffer = Some(chunk);
-                            }
-
-                            Some(Err(err)) => {
-                                error!(
-                                    "Response payload stream error: {:?}",
-                                    err.into()
-                                );
-
-                                return Poll::Ready(());
-                            }
-                        },
-                    }
-                }
-            }
-
-            ServiceResponseStateProj::SendErrorPayload(ref mut stream, ref mut body) => {
-                // TODO: de-dupe impl with SendPayload
-
-                loop {
-                    match this.buffer {
-                        Some(ref mut buffer) => match ready!(stream.poll_capacity(cx)) {
-                            None => return Poll::Ready(()),
-
-                            Some(Ok(cap)) => {
-                                let len = buffer.len();
-                                let bytes = buffer.split_to(cmp::min(cap, len));
-
-                                if let Err(e) = stream.send_data(bytes, false) {
-                                    warn!("{:?}", e);
-                                    return Poll::Ready(());
-                                } else if !buffer.is_empty() {
-                                    let cap = cmp::min(buffer.len(), CHUNK_SIZE);
-                                    stream.reserve_capacity(cap);
-                                } else {
-                                    this.buffer.take();
-                                }
-                            }
-
-                            Some(Err(e)) => {
-                                warn!("{:?}", e);
-                                return Poll::Ready(());
-                            }
-                        },
-
-                        None => match ready!(body.as_mut().poll_next(cx)) {
-                            None => {
-                                if let Err(e) = stream.send_data(Bytes::new(), true) {
-                                    warn!("{:?}", e);
-                                }
-                                return Poll::Ready(());
-                            }
-
-                            Some(Ok(chunk)) => {
-                                stream
-                                    .reserve_capacity(cmp::min(chunk.len(), CHUNK_SIZE));
-                                *this.buffer = Some(chunk);
-                            }
-
-                            Some(Err(err)) => {
-                                error!("Response payload stream error: {:?}", err);
-
-                                return Poll::Ready(());
-                            }
-                        },
-                    }
-                }
-            }
-        }
-    }
 }

actix-http/src/header/shared/content_encoding.rs

@@ -23,6 +23,9 @@ pub enum ContentEncoding {
     /// Gzip algorithm.
     Gzip,
 
+    // Zstd algorithm.
+    Zstd,
+
     /// Indicates the identity function (i.e. no compression, nor modification).
     Identity,
 }
@@ -41,6 +44,7 @@ impl ContentEncoding {
             ContentEncoding::Br => "br",
             ContentEncoding::Gzip => "gzip",
             ContentEncoding::Deflate => "deflate",
+            ContentEncoding::Zstd => "zstd",
             ContentEncoding::Identity | ContentEncoding::Auto => "identity",
         }
     }
@@ -53,6 +57,7 @@ impl ContentEncoding {
             ContentEncoding::Gzip => 1.0,
             ContentEncoding::Deflate => 0.9,
             ContentEncoding::Identity | ContentEncoding::Auto => 0.1,
+            ContentEncoding::Zstd => 0.0,
         }
     }
 }
@@ -81,6 +86,8 @@ impl From<&str> for ContentEncoding {
             ContentEncoding::Gzip
         } else if val.eq_ignore_ascii_case("deflate") {
             ContentEncoding::Deflate
+        } else if val.eq_ignore_ascii_case("zstd") {
+            ContentEncoding::Zstd
         } else {
             ContentEncoding::default()
         }

actix-web-codegen/tests/trybuild.rs

@@ -1,3 +1,4 @@
+#[rustversion::stable(1.46)] // MSRV
 #[test]
 fn compile_macros() {
     let t = trybuild::TestCases::new();
@@ -12,11 +13,3 @@ fn compile_macros() {
 
     t.pass("tests/trybuild/docstring-ok.rs");
 }
-
-// #[rustversion::not(nightly)]
-// fn skip_on_nightly(t: &trybuild::TestCases) {
-//
-// }
-
-// #[rustversion::nightly]
-// fn skip_on_nightly(_t: &trybuild::TestCases) {}

src/lib.rs

@@ -26,6 +26,7 @@
 //!
 //! * [Website & User Guide](https://actix.rs/)
 //! * [Examples Repository](https://github.com/actix/examples)
+//! * [Community Chat on Discord](https://discord.gg/NWpN5mmg3x)
 //! * [Community Chat on Gitter](https://gitter.im/actix/actix-web)
 //!
 //! To get started navigating the API docs, you may consider looking at the following pages first:

src/middleware/normalize.rs

@@ -189,6 +189,7 @@ mod tests {
     use super::*;
     use crate::{
         dev::ServiceRequest,
+        guard::fn_guard,
         test::{call_service, init_service, TestRequest},
         web, App, HttpResponse,
     };
@@ -199,37 +200,34 @@ mod tests {
             App::new()
                 .wrap(NormalizePath::default())
                 .service(web::resource("/").to(HttpResponse::Ok))
-                .service(web::resource("/v1/something").to(HttpResponse::Ok)),
+                .service(web::resource("/v1/something").to(HttpResponse::Ok))
+                .service(
+                    web::resource("/v2/something")
+                        .guard(fn_guard(|req| req.uri.query() == Some("query=test")))
+                        .to(HttpResponse::Ok),
+                ),
         )
         .await;
 
-        let req = TestRequest::with_uri("/").to_request();
+        let test_uris = vec![
+            "/",
+            "/?query=test",
+            "///",
+            "/v1//something",
+            "/v1//something////",
+            "//v1/something",
+            "//v1//////something",
+            "/v2//something?query=test",
+            "/v2//something////?query=test",
+            "//v2/something?query=test",
+            "//v2//////something?query=test",
+        ];
+
+        for uri in test_uris {
+            let req = TestRequest::with_uri(uri).to_request();
             let res = call_service(&app, req).await;
-        assert!(res.status().is_success());
+            assert!(res.status().is_success(), "Failed uri: {}", uri);
-
+        }
-        let req = TestRequest::with_uri("/?query=test").to_request();
-        let res = call_service(&app, req).await;
-        assert!(res.status().is_success());
-
-        let req = TestRequest::with_uri("///").to_request();
-        let res = call_service(&app, req).await;
-        assert!(res.status().is_success());
-
-        let req = TestRequest::with_uri("/v1//something////").to_request();
-        let res = call_service(&app, req).await;
-        assert!(res.status().is_success());
-
-        let req2 = TestRequest::with_uri("//v1/something").to_request();
-        let res2 = call_service(&app, req2).await;
-        assert!(res2.status().is_success());
-
-        let req3 = TestRequest::with_uri("//v1//////something").to_request();
-        let res3 = call_service(&app, req3).await;
-        assert!(res3.status().is_success());
-
-        let req4 = TestRequest::with_uri("/v1//something").to_request();
-        let res4 = call_service(&app, req4).await;
-        assert!(res4.status().is_success());
     }
 
     #[actix_rt::test]
@@ -238,38 +236,114 @@ mod tests {
             App::new()
                 .wrap(NormalizePath(TrailingSlash::Trim))
                 .service(web::resource("/").to(HttpResponse::Ok))
-                .service(web::resource("/v1/something").to(HttpResponse::Ok)),
+                .service(web::resource("/v1/something").to(HttpResponse::Ok))
+                .service(
+                    web::resource("/v2/something")
+                        .guard(fn_guard(|req| req.uri.query() == Some("query=test")))
+                        .to(HttpResponse::Ok),
+                ),
         )
         .await;
 
-        // root paths should still work
+        let test_uris = vec![
-        let req = TestRequest::with_uri("/").to_request();
+            "/",
+            "///",
+            "/v1/something",
+            "/v1/something/",
+            "/v1/something////",
+            "//v1//something",
+            "//v1//something//",
+            "/v2/something?query=test",
+            "/v2/something/?query=test",
+            "/v2/something////?query=test",
+            "//v2//something?query=test",
+            "//v2//something//?query=test",
+        ];
+
+        for uri in test_uris {
+            let req = TestRequest::with_uri(uri).to_request();
             let res = call_service(&app, req).await;
-        assert!(res.status().is_success());
+            assert!(res.status().is_success(), "Failed uri: {}", uri);
+        }
+    }
 
-        let req = TestRequest::with_uri("/?query=test").to_request();
+    #[actix_rt::test]
+    async fn trim_root_trailing_slashes_with_query() {
+        let app = init_service(
+            App::new().wrap(NormalizePath(TrailingSlash::Trim)).service(
+                web::resource("/")
+                    .guard(fn_guard(|req| req.uri.query() == Some("query=test")))
+                    .to(HttpResponse::Ok),
+            ),
+        )
+        .await;
+
+        let test_uris = vec!["/?query=test", "//?query=test", "///?query=test"];
+
+        for uri in test_uris {
+            let req = TestRequest::with_uri(uri).to_request();
             let res = call_service(&app, req).await;
-        assert!(res.status().is_success());
+            assert!(res.status().is_success(), "Failed uri: {}", uri);
+        }
+    }
 
-        let req = TestRequest::with_uri("///").to_request();
+    #[actix_rt::test]
+    async fn ensure_trailing_slash() {
+        let app = init_service(
+            App::new()
+                .wrap(NormalizePath(TrailingSlash::Always))
+                .service(web::resource("/").to(HttpResponse::Ok))
+                .service(web::resource("/v1/something/").to(HttpResponse::Ok))
+                .service(
+                    web::resource("/v2/something/")
+                        .guard(fn_guard(|req| req.uri.query() == Some("query=test")))
+                        .to(HttpResponse::Ok),
+                ),
+        )
+        .await;
+
+        let test_uris = vec![
+            "/",
+            "///",
+            "/v1/something",
+            "/v1/something/",
+            "/v1/something////",
+            "//v1//something",
+            "//v1//something//",
+            "/v2/something?query=test",
+            "/v2/something/?query=test",
+            "/v2/something////?query=test",
+            "//v2//something?query=test",
+            "//v2//something//?query=test",
+        ];
+
+        for uri in test_uris {
+            let req = TestRequest::with_uri(uri).to_request();
             let res = call_service(&app, req).await;
-        assert!(res.status().is_success());
+            assert!(res.status().is_success(), "Failed uri: {}", uri);
+        }
+    }
 
-        let req = TestRequest::with_uri("/v1/something////").to_request();
+    #[actix_rt::test]
+    async fn ensure_root_trailing_slash_with_query() {
+        let app = init_service(
+            App::new()
+                .wrap(NormalizePath(TrailingSlash::Always))
+                .service(
+                    web::resource("/")
+                        .guard(fn_guard(|req| req.uri.query() == Some("query=test")))
+                        .to(HttpResponse::Ok),
+                ),
+        )
+        .await;
+
+        let test_uris = vec!["/?query=test", "//?query=test", "///?query=test"];
+
+        for uri in test_uris {
+            let req = TestRequest::with_uri(uri).to_request();
             let res = call_service(&app, req).await;
-        assert!(res.status().is_success());
+            assert!(res.status().is_success(), "Failed uri: {}", uri);
-
+        }
-        let req2 = TestRequest::with_uri("/v1/something/").to_request();
-        let res2 = call_service(&app, req2).await;
-        assert!(res2.status().is_success());
-
-        let req3 = TestRequest::with_uri("//v1//something//").to_request();
-        let res3 = call_service(&app, req3).await;
-        assert!(res3.status().is_success());
-
-        let req4 = TestRequest::with_uri("//v1//something").to_request();
-        let res4 = call_service(&app, req4).await;
-        assert!(res4.status().is_success());
     }
 
     #[actix_rt::test]
@@ -279,7 +353,12 @@ mod tests {
                 .wrap(NormalizePath(TrailingSlash::MergeOnly))
                 .service(web::resource("/").to(HttpResponse::Ok))
                 .service(web::resource("/v1/something").to(HttpResponse::Ok))
-                .service(web::resource("/v1/").to(HttpResponse::Ok)),
+                .service(web::resource("/v1/").to(HttpResponse::Ok))
+                .service(
+                    web::resource("/v2/something")
+                        .guard(fn_guard(|req| req.uri.query() == Some("query=test")))
+                        .to(HttpResponse::Ok),
+                ),
         )
         .await;
 
@@ -295,12 +374,16 @@ mod tests {
             ("/v1////", true),
             ("//v1//", true),
             ("///v1", false),
+            ("/v2/something?query=test", true),
+            ("/v2/something/?query=test", false),
+            ("/v2/something//?query=test", false),
+            ("//v2//something?query=test", true),
         ];
 
-        for (path, success) in tests {
+        for (uri, success) in tests {
-            let req = TestRequest::with_uri(path).to_request();
+            let req = TestRequest::with_uri(uri).to_request();
             let res = call_service(&app, req).await;
-            assert_eq!(res.status().is_success(), success);
+            assert_eq!(res.status().is_success(), success, "Failed uri: {}", uri);
         }
     }
 
@@ -316,21 +399,18 @@ mod tests {
             .await
             .unwrap();
 
-        let req = TestRequest::with_uri("/v1//something////").to_srv_request();
+        let test_uris = vec![
+            "/v1//something////",
+            "///v1/something",
+            "//v1///something",
+            "/v1//something",
+        ];
+
+        for uri in test_uris {
+            let req = TestRequest::with_uri(uri).to_srv_request();
             let res = normalize.call(req).await.unwrap();
-        assert!(res.status().is_success());
+            assert!(res.status().is_success(), "Failed uri: {}", uri);
-
+        }
-        let req2 = TestRequest::with_uri("///v1/something").to_srv_request();
-        let res2 = normalize.call(req2).await.unwrap();
-        assert!(res2.status().is_success());
-
-        let req3 = TestRequest::with_uri("//v1///something").to_srv_request();
-        let res3 = normalize.call(req3).await.unwrap();
-        assert!(res3.status().is_success());
-
-        let req4 = TestRequest::with_uri("/v1//something").to_srv_request();
-        let res4 = normalize.call(req4).await.unwrap();
-        assert!(res4.status().is_success());
     }
 
     #[actix_rt::test]

tests/test_server.rs

@@ -29,6 +29,7 @@ use openssl::{
     x509::X509,
 };
 use rand::{distributions::Alphanumeric, Rng};
+use zstd::stream::{read::Decoder as ZstdDecoder, write::Encoder as ZstdEncoder};
 
 use actix_web::dev::BodyEncoding;
 use actix_web::middleware::{Compress, NormalizePath, TrailingSlash};
@@ -476,6 +477,125 @@ async fn test_body_brotli() {
     assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref()));
 }
 
+#[actix_rt::test]
+async fn test_body_zstd() {
+    let srv = actix_test::start_with(actix_test::config().h1(), || {
+        App::new()
+            .wrap(Compress::new(ContentEncoding::Zstd))
+            .service(web::resource("/").route(web::to(move || HttpResponse::Ok().body(STR))))
+    });
+
+    // client request
+    let mut response = srv
+        .get("/")
+        .append_header((ACCEPT_ENCODING, "zstd"))
+        .no_decompress()
+        .send()
+        .await
+        .unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+
+    // decode
+    let mut e = ZstdDecoder::new(&bytes[..]).unwrap();
+    let mut dec = Vec::new();
+    e.read_to_end(&mut dec).unwrap();
+    assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref()));
+}
+
+#[actix_rt::test]
+async fn test_body_zstd_streaming() {
+    let srv = actix_test::start_with(actix_test::config().h1(), || {
+        App::new()
+            .wrap(Compress::new(ContentEncoding::Zstd))
+            .service(web::resource("/").route(web::to(move || {
+                HttpResponse::Ok()
+                    .streaming(TestBody::new(Bytes::from_static(STR.as_ref()), 24))
+            })))
+    });
+
+    // client request
+    let mut response = srv
+        .get("/")
+        .append_header((ACCEPT_ENCODING, "zstd"))
+        .no_decompress()
+        .send()
+        .await
+        .unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+
+    // decode
+    let mut e = ZstdDecoder::new(&bytes[..]).unwrap();
+    let mut dec = Vec::new();
+    e.read_to_end(&mut dec).unwrap();
+    assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref()));
+}
+
+#[actix_rt::test]
+async fn test_zstd_encoding() {
+    let srv = actix_test::start_with(actix_test::config().h1(), || {
+        App::new().service(
+            web::resource("/").route(web::to(move |body: Bytes| HttpResponse::Ok().body(body))),
+        )
+    });
+
+    let mut e = ZstdEncoder::new(Vec::new(), 5).unwrap();
+    e.write_all(STR.as_ref()).unwrap();
+    let enc = e.finish().unwrap();
+
+    // client request
+    let request = srv
+        .post("/")
+        .append_header((CONTENT_ENCODING, "zstd"))
+        .send_body(enc.clone());
+    let mut response = request.await.unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().await.unwrap();
+    assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
+}
+
+#[actix_rt::test]
+async fn test_zstd_encoding_large() {
+    let data = rand::thread_rng()
+        .sample_iter(&Alphanumeric)
+        .take(320_000)
+        .map(char::from)
+        .collect::<String>();
+
+    let srv = actix_test::start_with(actix_test::config().h1(), || {
+        App::new().service(
+            web::resource("/")
+                .app_data(web::PayloadConfig::new(320_000))
+                .route(web::to(move |body: Bytes| {
+                    HttpResponse::Ok().streaming(TestBody::new(body, 10240))
+                })),
+        )
+    });
+
+    let mut e = ZstdEncoder::new(Vec::new(), 5).unwrap();
+    e.write_all(data.as_ref()).unwrap();
+    let enc = e.finish().unwrap();
+
+    // client request
+    let request = srv
+        .post("/")
+        .append_header((CONTENT_ENCODING, "zstd"))
+        .send_body(enc.clone());
+    let mut response = request.await.unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = response.body().limit(320_000).await.unwrap();
+    assert_eq!(bytes, Bytes::from(data));
+}
+
 #[actix_rt::test]
 async fn test_encoding() {
     let srv = actix_test::start_with(actix_test::config().h1(), || {