Replace brotli with brotli2 #1224

Cargo.toml

@@ -93,11 +93,11 @@ open-ssl = { version="0.10", package = "openssl", optional = true }
 rust-tls = { version = "0.16.0", package = "rustls", optional = true }
 
 [dev-dependencies]
-actix = "0.9.0-alpha.1"
+actix = "0.9.0-alpha.2"
 rand = "0.7"
 env_logger = "0.6"
 serde_derive = "1.0"
-brotli = "3.3.0"
+brotli2 = "0.3.2"
 flate2 = "1.0.13"
 
 [profile.release]

actix-http/CHANGES.md

@@ -1,11 +1,13 @@
 # Changes
 
-## [1.0.1] - 2019-12-xx
+## [1.0.1] - 2019-12-20
 
 ### Fixed
 
 * Poll upgrade service's readiness from HTTP service handlers
 
+* Replace brotli with brotli2 #1224
+
 ## [1.0.0] - 2019-12-13
 
 ### Added

actix-http/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-http"
-version = "1.0.0"
+version = "1.0.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix http primitives"
 readme = "README.md"
@@ -31,7 +31,7 @@ openssl = ["actix-tls/openssl", "actix-connect/openssl"]
 rustls = ["actix-tls/rustls", "actix-connect/rustls"]
 
 # enable compressison support
-compress = ["flate2", "brotli"]
+compress = ["flate2", "brotli2"]
 
 # failure integration. actix does not use failure anymore
 failure = ["fail-ure"]
@@ -83,7 +83,7 @@ time = "0.1.42"
 ring = { version = "0.16.9", optional = true }
 
 # compression
-brotli = { version = "3.3.0", optional = true }
+brotli2 = { version="0.3.2", optional = true }
 flate2 = { version = "1.0.13", optional = true }
 
 # optional deps

actix-http/src/encoding/decoder.rs

@@ -4,7 +4,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use actix_threadpool::{run, CpuFuture};
-use brotli::DecompressorWriter;
+use brotli2::write::BrotliDecoder;
 use bytes::Bytes;
 use flate2::write::{GzDecoder, ZlibDecoder};
 use futures_core::{ready, Stream};
@@ -31,7 +31,7 @@ where
     pub fn new(stream: S, encoding: ContentEncoding) -> Decoder<S> {
         let decoder = match encoding {
             ContentEncoding::Br => Some(ContentDecoder::Br(Box::new(
-                DecompressorWriter::new(Writer::new(), 0),
+                BrotliDecoder::new(Writer::new()),
             ))),
             ContentEncoding::Deflate => Some(ContentDecoder::Deflate(Box::new(
                 ZlibDecoder::new(Writer::new()),
@@ -137,7 +137,7 @@ where
 enum ContentDecoder {
     Deflate(Box<ZlibDecoder<Writer>>),
     Gzip(Box<GzDecoder<Writer>>),
-    Br(Box<DecompressorWriter<Writer>>),
+    Br(Box<BrotliDecoder<Writer>>),
 }
 
 impl ContentDecoder {

actix-http/src/encoding/encoder.rs

@@ -5,7 +5,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use actix_threadpool::{run, CpuFuture};
-use brotli::CompressorWriter;
+use brotli2::write::BrotliEncoder;
 use bytes::Bytes;
 use flate2::write::{GzEncoder, ZlibEncoder};
 use futures_core::ready;
@@ -17,7 +17,7 @@ use crate::{Error, ResponseHead};
 
 use super::Writer;
 
-const INPLACE: usize = 2049;
+const INPLACE: usize = 1024;
 
 pub struct Encoder<B> {
     eof: bool,
@@ -174,7 +174,7 @@ fn update_head(encoding: ContentEncoding, head: &mut ResponseHead) {
 enum ContentEncoder {
     Deflate(ZlibEncoder<Writer>),
     Gzip(GzEncoder<Writer>),
-    Br(Box<CompressorWriter<Writer>>),
+    Br(BrotliEncoder<Writer>),
 }
 
 impl ContentEncoder {
@@ -188,9 +188,9 @@ impl ContentEncoder {
                 Writer::new(),
                 flate2::Compression::fast(),
             ))),
-            ContentEncoding::Br => Some(ContentEncoder::Br(Box::new(
+            ContentEncoding::Br => {
-                CompressorWriter::new(Writer::new(), 0, 3, 0),
+                Some(ContentEncoder::Br(BrotliEncoder::new(Writer::new(), 3)))
-            ))),
+            }
             _ => None,
         }
     }
@@ -198,12 +198,7 @@ impl ContentEncoder {
     #[inline]
     pub(crate) fn take(&mut self) -> Bytes {
         match *self {
-            ContentEncoder::Br(ref mut encoder) => {
+            ContentEncoder::Br(ref mut encoder) => encoder.get_mut().take(),
-                let mut encoder_new =
-                    Box::new(CompressorWriter::new(Writer::new(), 0, 3, 0));
-                std::mem::swap(encoder, &mut encoder_new);
-                encoder_new.into_inner().freeze()
-            }
             ContentEncoder::Deflate(ref mut encoder) => encoder.get_mut().take(),
             ContentEncoder::Gzip(ref mut encoder) => encoder.get_mut().take(),
         }
@@ -211,7 +206,10 @@ impl ContentEncoder {
 
     fn finish(self) -> Result<Bytes, io::Error> {
         match self {
-            ContentEncoder::Br(encoder) => Ok(encoder.into_inner().buf.freeze()),
+            ContentEncoder::Br(encoder) => match encoder.finish() {
+                Ok(writer) => Ok(writer.buf.freeze()),
+                Err(err) => Err(err),
+            },
             ContentEncoder::Gzip(encoder) => match encoder.finish() {
                 Ok(writer) => Ok(writer.buf.freeze()),
                 Err(err) => Err(err),

actix-http/src/encoding/mod.rs

@@ -19,12 +19,10 @@ impl Writer {
             buf: BytesMut::with_capacity(8192),
         }
     }
+
     fn take(&mut self) -> Bytes {
         self.buf.split().freeze()
     }
-    fn freeze(self) -> Bytes {
-        self.buf.freeze()
-    }
 }
 
 impl io::Write for Writer {
@@ -32,6 +30,7 @@ impl io::Write for Writer {
         self.buf.extend_from_slice(buf);
         Ok(buf.len())
     }
+
     fn flush(&mut self) -> io::Result<()> {
         Ok(())
     }

actix-http/src/error.rs

@@ -6,7 +6,9 @@ use std::str::Utf8Error;
 use std::string::FromUtf8Error;
 use std::{fmt, io, result};
 
+use actix_codec::{Decoder, Encoder};
 pub use actix_threadpool::BlockingError;
+use actix_utils::framed::DispatcherError as FramedDispatcherError;
 use actix_utils::timeout::TimeoutError;
 use bytes::BytesMut;
 use derive_more::{Display, From};
@@ -463,6 +465,14 @@ impl ResponseError for ContentTypeError {
     }
 }
 
+impl<E, U: Encoder + Decoder> ResponseError for FramedDispatcherError<E, U>
+where
+    E: fmt::Debug + fmt::Display,
+    <U as Encoder>::Error: fmt::Debug,
+    <U as Decoder>::Error: fmt::Debug,
+{
+}
+
 /// Helper type that can wrap any error and generate custom response.
 ///
 /// In following example any `io::Error` will be converted into "BAD REQUEST"

actix-http/src/h1/service.rs

@@ -165,7 +165,7 @@ mod rustls {
             Request = (Request, Framed<TlsStream<TcpStream>, Codec>),
             Response = (),
         >,
-        U::Error: fmt::Display,
+        U::Error: fmt::Display + Into<Error>,
         U::InitError: fmt::Debug,
     {
         /// Create rustls based service

actix-http/src/service.rs

@@ -276,7 +276,7 @@ mod rustls {
             Request = (Request, Framed<TlsStream<TcpStream>, h1::Codec>),
             Response = (),
         >,
-        U::Error: fmt::Display,
+        U::Error: fmt::Display + Into<Error>,
         U::InitError: fmt::Debug,
         <U::Service as Service>::Future: 'static,
     {

awc/Cargo.toml

@@ -61,7 +61,7 @@ actix-http-test = { version = "1.0.0", features=["openssl"] }
 actix-utils = "1.0.3"
 actix-server = "1.0.0"
 actix-tls = { version = "1.0.0", features=["openssl", "rustls"] }
-brotli = "3.3.0"
+brotli2 = "0.3.2"
 flate2 = "1.0.13"
 futures = "0.3.1"
 env_logger = "0.6"

awc/tests/test_client.rs

@@ -4,7 +4,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
-use brotli::CompressorWriter;
+use brotli2::write::BrotliEncoder;
 use bytes::Bytes;
 use flate2::read::GzDecoder;
 use flate2::write::GzEncoder;
@@ -500,9 +500,9 @@ async fn test_client_gzip_encoding_large_random() {
 async fn test_client_brotli_encoding() {
     let srv = test::start(|| {
         App::new().service(web::resource("/").route(web::to(|data: Bytes| {
-            let mut e = CompressorWriter::new(Vec::new(), 0, 5, 0);
+            let mut e = BrotliEncoder::new(Vec::new(), 5);
             e.write_all(&data).unwrap();
-            let data = e.into_inner();
+            let data = e.finish().unwrap();
             HttpResponse::Ok()
                 .header("content-encoding", "br")
                 .body(data)
@@ -527,9 +527,9 @@ async fn test_client_brotli_encoding_large_random() {
 
     let srv = test::start(|| {
         App::new().service(web::resource("/").route(web::to(|data: Bytes| {
-            let mut e = CompressorWriter::new(Vec::new(), 0, 5, 0);
+            let mut e = BrotliEncoder::new(Vec::new(), 5);
             e.write_all(&data).unwrap();
-            let data = e.into_inner();
+            let data = e.finish().unwrap();
             HttpResponse::Ok()
                 .header("content-encoding", "br")
                 .body(data)

tests/test_server.rs

@@ -1,15 +1,17 @@
 use std::io::{Read, Write};
+use std::pin::Pin;
+use std::task::{Context, Poll};
 
 use actix_http::http::header::{
     ContentEncoding, ACCEPT_ENCODING, CONTENT_ENCODING, CONTENT_LENGTH,
     TRANSFER_ENCODING,
 };
-use brotli::{CompressorWriter, DecompressorWriter};
+use brotli2::write::{BrotliDecoder, BrotliEncoder};
 use bytes::Bytes;
 use flate2::read::GzDecoder;
 use flate2::write::{GzEncoder, ZlibDecoder, ZlibEncoder};
 use flate2::Compression;
-use futures::{future::ok, stream::once};
+use futures::{ready, Future};
 use rand::{distributions::Alphanumeric, Rng};
 
 use actix_web::dev::BodyEncoding;
@@ -38,6 +40,42 @@ const STR: &str = "Hello World Hello World Hello World Hello World Hello World \
                    Hello World Hello World Hello World Hello World Hello World \
                    Hello World Hello World Hello World Hello World Hello World";
 
+struct TestBody {
+    data: Bytes,
+    chunk_size: usize,
+    delay: actix_rt::time::Delay,
+}
+
+impl TestBody {
+    fn new(data: Bytes, chunk_size: usize) -> Self {
+        TestBody {
+            data,
+            chunk_size,
+            delay: actix_rt::time::delay_for(std::time::Duration::from_millis(10)),
+        }
+    }
+}
+
+impl futures::Stream for TestBody {
+    type Item = Result<Bytes, Error>;
+
+    fn poll_next(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        ready!(Pin::new(&mut self.delay).poll(cx));
+
+        self.delay = actix_rt::time::delay_for(std::time::Duration::from_millis(10));
+        let chunk_size = std::cmp::min(self.chunk_size, self.data.len());
+        let chunk = self.data.split_to(chunk_size);
+        if chunk.is_empty() {
+            Poll::Ready(None)
+        } else {
+            Poll::Ready(Some(Ok(chunk)))
+        }
+    }
+}
+
 #[actix_rt::test]
 async fn test_body() {
     let srv = test::start(|| {
@@ -248,7 +286,7 @@ async fn test_body_chunked_implicit() {
             .wrap(Compress::new(ContentEncoding::Gzip))
             .service(web::resource("/").route(web::get().to(move || {
                 HttpResponse::Ok()
-                    .streaming(once(ok::<_, Error>(Bytes::from_static(STR.as_ref()))))
+                    .streaming(TestBody::new(Bytes::from_static(STR.as_ref()), 24))
             })))
     });
 
@@ -281,7 +319,7 @@ async fn test_body_br_streaming() {
         App::new().wrap(Compress::new(ContentEncoding::Br)).service(
             web::resource("/").route(web::to(move || {
                 HttpResponse::Ok()
-                    .streaming(once(ok::<_, Error>(Bytes::from_static(STR.as_ref()))))
+                    .streaming(TestBody::new(Bytes::from_static(STR.as_ref()), 24))
             })),
         )
     });
@@ -297,11 +335,13 @@ async fn test_body_br_streaming() {
 
     // read response
     let bytes = response.body().await.unwrap();
+    println!("TEST: {:?}", bytes.len());
 
     // decode br
-    let mut e = DecompressorWriter::new(Vec::with_capacity(2048), 0);
+    let mut e = BrotliDecoder::new(Vec::with_capacity(2048));
     e.write_all(bytes.as_ref()).unwrap();
-    let dec = e.into_inner().unwrap();
+    let dec = e.finish().unwrap();
+    println!("T: {:?}", Bytes::copy_from_slice(&dec));
     assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref()));
 }
 
@@ -333,7 +373,7 @@ async fn test_no_chunking() {
             HttpResponse::Ok()
                 .no_chunking()
                 .content_length(STR.len() as u64)
-                .streaming(once(ok::<_, Error>(Bytes::from_static(STR.as_ref()))))
+                .streaming(TestBody::new(Bytes::from_static(STR.as_ref()), 24))
         })))
     });
 
@@ -397,9 +437,9 @@ async fn test_body_brotli() {
     let bytes = response.body().await.unwrap();
 
     // decode brotli
-    let mut e = DecompressorWriter::new(Vec::with_capacity(2048), 0);
+    let mut e = BrotliDecoder::new(Vec::with_capacity(2048));
     e.write_all(bytes.as_ref()).unwrap();
-    let dec = e.into_inner().unwrap();
+    let dec = e.finish().unwrap();
     assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref()));
 }
 
@@ -608,9 +648,9 @@ async fn test_brotli_encoding() {
         )
     });
 
-    let mut e = CompressorWriter::new(Vec::new(), 0, 3, 0);
+    let mut e = BrotliEncoder::new(Vec::new(), 5);
     e.write_all(STR.as_ref()).unwrap();
-    let enc = e.into_inner();
+    let enc = e.finish().unwrap();
 
     // client request
     let request = srv
@@ -627,17 +667,24 @@ async fn test_brotli_encoding() {
 
 #[actix_rt::test]
 async fn test_brotli_encoding_large() {
-    let data = STR.repeat(10);
+    let data = rand::thread_rng()
+        .sample_iter(&Alphanumeric)
+        .take(320_000)
+        .collect::<String>();
+
     let srv = test::start_with(test::config().h1(), || {
         App::new().service(
             web::resource("/")
-                .route(web::to(move |body: Bytes| HttpResponse::Ok().body(body))),
+                .data(web::PayloadConfig::new(320_000))
+                .route(web::to(move |body: Bytes| {
+                    HttpResponse::Ok().streaming(TestBody::new(body, 10240))
+                })),
         )
     });
 
-    let mut e = CompressorWriter::new(Vec::new(), 0, 3, 0);
+    let mut e = BrotliEncoder::new(Vec::new(), 5);
     e.write_all(data.as_ref()).unwrap();
-    let enc = e.into_inner();
+    let enc = e.finish().unwrap();
 
     // client request
     let request = srv
@@ -648,7 +695,7 @@ async fn test_brotli_encoding_large() {
     assert!(response.status().is_success());
 
     // read response
-    let bytes = response.body().await.unwrap();
+    let bytes = response.body().limit(320_000).await.unwrap();
     assert_eq!(bytes, Bytes::from(data));
 }
 
@@ -675,9 +722,9 @@ async fn test_brotli_encoding_large_openssl() {
     });
 
     // body
-    let mut e = CompressorWriter::new(Vec::new(), 0, 3, 0);
+    let mut e = BrotliEncoder::new(Vec::new(), 3);
     e.write_all(data.as_ref()).unwrap();
-    let enc = e.into_inner();
+    let enc = e.finish().unwrap();
 
     // client request
     let mut response = srv
@@ -731,7 +778,7 @@ async fn test_reading_deflate_encoding_large_random_rustls() {
     let req = srv
         .post("/")
         .header(actix_web::http::header::CONTENT_ENCODING, "deflate")
-        .send_body(enc);
+        .send_stream(TestBody::new(Bytes::from(enc), 1024));
 
     let mut response = req.await.unwrap();
     assert!(response.status().is_success());
@@ -742,93 +789,6 @@ async fn test_reading_deflate_encoding_large_random_rustls() {
     assert_eq!(bytes, Bytes::from(data));
 }
 
-// #[cfg(all(feature = "tls", feature = "ssl"))]
-// #[test]
-// fn test_reading_deflate_encoding_large_random_nativetls() {
-//     use native_tls::{Identity, TlsAcceptor};
-//     use openssl::ssl::{
-//         SslAcceptor, SslConnector, SslFiletype, SslMethod, SslVerifyMode,
-//     };
-//     use std::fs::File;
-//     use std::sync::mpsc;
-
-//     use actix::{Actor, System};
-//     let (tx, rx) = mpsc::channel();
-
-//     // load ssl keys
-//     let mut file = File::open("tests/identity.pfx").unwrap();
-//     let mut identity = vec![];
-//     file.read_to_end(&mut identity).unwrap();
-//     let identity = Identity::from_pkcs12(&identity, "1").unwrap();
-//     let acceptor = TlsAcceptor::new(identity).unwrap();
-
-//     // load ssl keys
-//     let mut builder = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
-//     builder
-//         .set_private_key_file("tests/key.pem", SslFiletype::PEM)
-//         .unwrap();
-//     builder
-//         .set_certificate_chain_file("tests/cert.pem")
-//         .unwrap();
-
-//     let data = rand::thread_rng()
-//         .sample_iter(&Alphanumeric)
-//         .take(160_000)
-//         .collect::<String>();
-
-//     let addr = test::TestServer::unused_addr();
-//     thread::spawn(move || {
-//         System::run(move || {
-//             server::new(|| {
-//                 App::new().handler("/", |req: &HttpRequest| {
-//                     req.body()
-//                         .and_then(|bytes: Bytes| {
-//                             Ok(HttpResponse::Ok()
-//                                 .content_encoding(http::ContentEncoding::Identity)
-//                                 .body(bytes))
-//                         })
-//                         .responder()
-//                 })
-//             })
-//             .bind_tls(addr, acceptor)
-//             .unwrap()
-//             .start();
-//             let _ = tx.send(System::current());
-//         });
-//     });
-//     let sys = rx.recv().unwrap();
-
-//     let mut rt = System::new("test");
-
-//     // client connector
-//     let mut builder = SslConnector::builder(SslMethod::tls()).unwrap();
-//     builder.set_verify(SslVerifyMode::NONE);
-//     let conn = client::ClientConnector::with_connector(builder.build()).start();
-
-//     // encode data
-//     let mut e = ZlibEncoder::new(Vec::new(), Compression::default());
-//     e.write_all(data.as_ref()).unwrap();
-//     let enc = e.finish().unwrap();
-
-//     // client request
-//     let request = client::ClientRequest::build()
-//         .uri(format!("https://{}/", addr))
-//         .method(http::Method::POST)
-//         .header(http::header::CONTENT_ENCODING, "deflate")
-//         .with_connector(conn)
-//         .body(enc)
-//         .unwrap();
-//     let response = rt.block_on(request.send()).unwrap();
-//     assert!(response.status().is_success());
-
-//     // read response
-//     let bytes = rt.block_on(response.body()).unwrap();
-//     assert_eq!(bytes.len(), data.len());
-//     assert_eq!(bytes, Bytes::from(data));
-
-//     let _ = sys.stop();
-// }
-
 // #[test]
 // fn test_server_cookies() {
 //     use actix_web::http;