diff --git a/Cargo.toml b/Cargo.toml index 610ffb2b14b..0fab5d58caa 100644 --- a/Cargo.toml +++ b/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] diff --git a/actix-http/CHANGES.md b/actix-http/CHANGES.md index eb8620cf8e3..1c8e4f0533f 100644 --- a/actix-http/CHANGES.md +++ b/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 diff --git a/actix-http/Cargo.toml b/actix-http/Cargo.toml index 6669c7932a6..8512b250139 100644 --- a/actix-http/Cargo.toml +++ b/actix-http/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "actix-http" -version = "1.0.0" +version = "1.0.1" authors = ["Nikolay Kim "] 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 diff --git a/actix-http/src/encoding/decoder.rs b/actix-http/src/encoding/decoder.rs index cdc4699d5e8..b6043585915 100644 --- a/actix-http/src/encoding/decoder.rs +++ b/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 { 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>), Gzip(Box>), - Br(Box>), + Br(Box>), } impl ContentDecoder { diff --git a/actix-http/src/encoding/encoder.rs b/actix-http/src/encoding/encoder.rs index 6ec122fa00f..ca04845ab41 100644 --- a/actix-http/src/encoding/encoder.rs +++ b/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 { eof: bool, @@ -174,7 +174,7 @@ fn update_head(encoding: ContentEncoding, head: &mut ResponseHead) { enum ContentEncoder { Deflate(ZlibEncoder), Gzip(GzEncoder), - Br(Box>), + Br(BrotliEncoder), } impl ContentEncoder { @@ -188,9 +188,9 @@ impl ContentEncoder { Writer::new(), flate2::Compression::fast(), ))), - ContentEncoding::Br => Some(ContentEncoder::Br(Box::new( - CompressorWriter::new(Writer::new(), 0, 3, 0), - ))), + ContentEncoding::Br => { + 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) => { - 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::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(), } @@ -211,7 +206,10 @@ impl ContentEncoder { fn finish(self) -> Result { 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), diff --git a/actix-http/src/encoding/mod.rs b/actix-http/src/encoding/mod.rs index 48cf8325289..9eaf4104ec9 100644 --- a/actix-http/src/encoding/mod.rs +++ b/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(()) } diff --git a/actix-http/src/error.rs b/actix-http/src/error.rs index d252a0bb496..fd0fe927f65 100644 --- a/actix-http/src/error.rs +++ b/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 ResponseError for FramedDispatcherError +where + E: fmt::Debug + fmt::Display, + ::Error: fmt::Debug, + ::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" diff --git a/actix-http/src/h1/service.rs b/actix-http/src/h1/service.rs index 22f9d03b718..69c8fc55cf8 100644 --- a/actix-http/src/h1/service.rs +++ b/actix-http/src/h1/service.rs @@ -165,7 +165,7 @@ mod rustls { Request = (Request, Framed, Codec>), Response = (), >, - U::Error: fmt::Display, + U::Error: fmt::Display + Into, U::InitError: fmt::Debug, { /// Create rustls based service diff --git a/actix-http/src/service.rs b/actix-http/src/service.rs index 457c5ca1f0e..82618289bbc 100644 --- a/actix-http/src/service.rs +++ b/actix-http/src/service.rs @@ -276,7 +276,7 @@ mod rustls { Request = (Request, Framed, h1::Codec>), Response = (), >, - U::Error: fmt::Display, + U::Error: fmt::Display + Into, U::InitError: fmt::Debug, ::Future: 'static, { diff --git a/awc/Cargo.toml b/awc/Cargo.toml index 863b7a1e4db..d1eaa7f699f 100644 --- a/awc/Cargo.toml +++ b/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" diff --git a/awc/tests/test_client.rs b/awc/tests/test_client.rs index 48c23cfa275..f8fed5bfd7d 100644 --- a/awc/tests/test_client.rs +++ b/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) diff --git a/tests/test_server.rs b/tests/test_server.rs index 01137522166..5019157e285 100644 --- a/tests/test_server.rs +++ b/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; + + fn poll_next( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + 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::(); + 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::(); - -// 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;