actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2025-06-25 22:49:21 +02:00
Code Issues Releases Wiki Activity

add content-encoding decompression

Browse Source
This commit is contained in:
Nikolay Kim
2019-03-26 15:14:32 -07:00
parent 9451ba71f4
commit 1904b01fc0
16 changed files with 780 additions and 538 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/app.rs
View File

@ -193,10 +193,10 @@ where
}
/// Register a request modifier. It can modify any request parameters
/// including payload stream type.
/// including request payload type.
pub fn chain<C, F, P1>(
self,
chain: C,
chain: F,
) -> App<
P1,
impl NewService<

2
src/app_service.rs
View File

@ -380,7 +380,7 @@ impl<P> Service for AppRouting<P> {
} else if let Some(ref mut default) = self.default {
Either::A(default.call(req))
} else {
let req = req.into_request();
let req = req.into_parts().0;
Either::B(ok(ServiceResponse::new(req, Response::NotFound().finish())))
}
}

278
src/middleware/compress.rs
View File

@ -1,25 +1,14 @@
/// `Middleware` for compressing response body.
use std::io::Write;
//! `Middleware` for compressing response body.
use std::cmp;
use std::marker::PhantomData;
use std::str::FromStr;
use std::{cmp, fmt, io};
use actix_http::body::{Body, BodyLength, MessageBody, ResponseBody};
use actix_http::http::header::{
ContentEncoding, HeaderValue, ACCEPT_ENCODING, CONTENT_ENCODING,
};
use actix_http::http::{HttpTryFrom, StatusCode};
use actix_http::{Error, Head, ResponseHead};
use actix_http::body::MessageBody;
use actix_http::encoding::Encoder;
use actix_http::http::header::{ContentEncoding, ACCEPT_ENCODING};
use actix_service::{Service, Transform};
use bytes::{Bytes, BytesMut};
use futures::future::{ok, FutureResult};
use futures::{Async, Future, Poll};
use log::trace;
#[cfg(feature = "brotli")]
use brotli2::write::BrotliEncoder;
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
use flate2::write::{GzEncoder, ZlibEncoder};
use crate::service::{ServiceRequest, ServiceResponse};
@ -130,266 +119,11 @@ where
let resp = futures::try_ready!(self.fut.poll());
Ok(Async::Ready(resp.map_body(move |head, body| {
Encoder::body(self.encoding, head, body)
Encoder::response(self.encoding, head, body)
})))
}
}
enum EncoderBody<B> {
Body(B),
Other(Box<dyn MessageBody>),
}
pub struct Encoder<B> {
body: EncoderBody<B>,
encoder: Option<ContentEncoder>,
}
impl<B: MessageBody> MessageBody for Encoder<B> {
fn length(&self) -> BodyLength {
if self.encoder.is_none() {
match self.body {
EncoderBody::Body(ref b) => b.length(),
EncoderBody::Other(ref b) => b.length(),
}
} else {
BodyLength::Stream
}
}
fn poll_next(&mut self) -> Poll<Option<Bytes>, Error> {
loop {
let result = match self.body {
EncoderBody::Body(ref mut b) => b.poll_next()?,
EncoderBody::Other(ref mut b) => b.poll_next()?,
};
match result {
Async::NotReady => return Ok(Async::NotReady),
Async::Ready(Some(chunk)) => {
if let Some(ref mut encoder) = self.encoder {
if encoder.write(&chunk)? {
return Ok(Async::Ready(Some(encoder.take())));
}
} else {
return Ok(Async::Ready(Some(chunk)));
}
}
Async::Ready(None) => {
if let Some(encoder) = self.encoder.take() {
let chunk = encoder.finish()?;
if chunk.is_empty() {
return Ok(Async::Ready(None));
} else {
return Ok(Async::Ready(Some(chunk)));
}
} else {
return Ok(Async::Ready(None));
}
}
}
}
}
}
fn update_head(encoding: ContentEncoding, head: &mut ResponseHead) {
head.headers_mut().insert(
CONTENT_ENCODING,
HeaderValue::try_from(Bytes::from_static(encoding.as_str().as_bytes())).unwrap(),
);
}
impl<B: MessageBody> Encoder<B> {
fn body(
encoding: ContentEncoding,
head: &mut ResponseHead,
body: ResponseBody<B>,
) -> ResponseBody<Encoder<B>> {
let has_ce = head.headers().contains_key(CONTENT_ENCODING);
match body {
ResponseBody::Other(b) => match b {
Body::None => ResponseBody::Other(Body::None),
Body::Empty => ResponseBody::Other(Body::Empty),
Body::Bytes(buf) => {
if !(has_ce
|| encoding == ContentEncoding::Identity
|| encoding == ContentEncoding::Auto)
{
let mut enc = ContentEncoder::encoder(encoding).unwrap();
// TODO return error!
let _ = enc.write(buf.as_ref());
let body = enc.finish().unwrap();
update_head(encoding, head);
ResponseBody::Other(Body::Bytes(body))
} else {
ResponseBody::Other(Body::Bytes(buf))
}
}
Body::Message(stream) => {
if has_ce || head.status == StatusCode::SWITCHING_PROTOCOLS {
ResponseBody::Body(Encoder {
body: EncoderBody::Other(stream),
encoder: None,
})
} else {
update_head(encoding, head);
head.no_chunking = false;
ResponseBody::Body(Encoder {
body: EncoderBody::Other(stream),
encoder: ContentEncoder::encoder(encoding),
})
}
}
},
ResponseBody::Body(stream) => {
if has_ce || head.status == StatusCode::SWITCHING_PROTOCOLS {
ResponseBody::Body(Encoder {
body: EncoderBody::Body(stream),
encoder: None,
})
} else {
update_head(encoding, head);
head.no_chunking = false;
ResponseBody::Body(Encoder {
body: EncoderBody::Body(stream),
encoder: ContentEncoder::encoder(encoding),
})
}
}
}
}
}
pub(crate) struct Writer {
buf: BytesMut,
}
impl Writer {
fn new() -> Writer {
Writer {
buf: BytesMut::with_capacity(8192),
}
}
fn take(&mut self) -> Bytes {
self.buf.take().freeze()
}
}
impl io::Write for Writer {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
self.buf.extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
pub(crate) enum ContentEncoder {
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
Deflate(ZlibEncoder<Writer>),
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
Gzip(GzEncoder<Writer>),
#[cfg(feature = "brotli")]
Br(BrotliEncoder<Writer>),
}
impl fmt::Debug for ContentEncoder {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match *self {
#[cfg(feature = "brotli")]
ContentEncoder::Br(_) => writeln!(f, "ContentEncoder(Brotli)"),
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
ContentEncoder::Deflate(_) => writeln!(f, "ContentEncoder(Deflate)"),
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
ContentEncoder::Gzip(_) => writeln!(f, "ContentEncoder(Gzip)"),
}
}
}
impl ContentEncoder {
fn encoder(encoding: ContentEncoding) -> Option<Self> {
match encoding {
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
ContentEncoding::Deflate => Some(ContentEncoder::Deflate(ZlibEncoder::new(
Writer::new(),
flate2::Compression::fast(),
))),
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
ContentEncoding::Gzip => Some(ContentEncoder::Gzip(GzEncoder::new(
Writer::new(),
flate2::Compression::fast(),
))),
#[cfg(feature = "brotli")]
ContentEncoding::Br => {
Some(ContentEncoder::Br(BrotliEncoder::new(Writer::new(), 3)))
}
_ => None,
}
}
#[inline]
pub(crate) fn take(&mut self) -> Bytes {
match *self {
#[cfg(feature = "brotli")]
ContentEncoder::Br(ref mut encoder) => encoder.get_mut().take(),
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
ContentEncoder::Deflate(ref mut encoder) => encoder.get_mut().take(),
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
ContentEncoder::Gzip(ref mut encoder) => encoder.get_mut().take(),
}
}
fn finish(self) -> Result<Bytes, io::Error> {
match self {
#[cfg(feature = "brotli")]
ContentEncoder::Br(encoder) => match encoder.finish() {
Ok(writer) => Ok(writer.buf.freeze()),
Err(err) => Err(err),
},
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
ContentEncoder::Gzip(encoder) => match encoder.finish() {
Ok(writer) => Ok(writer.buf.freeze()),
Err(err) => Err(err),
},
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
ContentEncoder::Deflate(encoder) => match encoder.finish() {
Ok(writer) => Ok(writer.buf.freeze()),
Err(err) => Err(err),
},
}
}
fn write(&mut self, data: &[u8]) -> Result<bool, io::Error> {
match *self {
#[cfg(feature = "brotli")]
ContentEncoder::Br(ref mut encoder) => match encoder.write_all(data) {
Ok(_) => Ok(!encoder.get_ref().buf.is_empty()),
Err(err) => {
trace!("Error decoding br encoding: {}", err);
Err(err)
}
},
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
ContentEncoder::Gzip(ref mut encoder) => match encoder.write_all(data) {
Ok(_) => Ok(!encoder.get_ref().buf.is_empty()),
Err(err) => {
trace!("Error decoding gzip encoding: {}", err);
Err(err)
}
},
#[cfg(any(feature = "flate2-c", feature = "flate2-rust"))]
ContentEncoder::Deflate(ref mut encoder) => match encoder.write_all(data) {
Ok(_) => Ok(!encoder.get_ref().buf.is_empty()),
Err(err) => {
trace!("Error decoding deflate encoding: {}", err);
Err(err)
}
},
}
}
}
struct AcceptEncoding {
encoding: ContentEncoding,
quality: f64,

60
src/middleware/decompress.rs Normal file
View File

@ -0,0 +1,60 @@
//! Chain service for decompressing request payload.
use std::marker::PhantomData;
use actix_http::encoding::Decoder;
use actix_service::{NewService, Service};
use bytes::Bytes;
use futures::future::{ok, FutureResult};
use futures::{Async, Poll, Stream};
use crate::dev::Payload;
use crate::error::{Error, PayloadError};
use crate::service::ServiceRequest;
use crate::HttpMessage;
pub struct Decompress<P>(PhantomData<P>);
impl<P> Decompress<P>
where
P: Stream<Item = Bytes, Error = PayloadError>,
{
pub fn new() -> Self {
Decompress(PhantomData)
}
}
impl<P> NewService for Decompress<P>
where
P: Stream<Item = Bytes, Error = PayloadError>,
{
type Request = ServiceRequest<P>;
type Response = ServiceRequest<Decoder<Payload<P>>>;
type Error = Error;
type InitError = ();
type Service = Decompress<P>;
type Future = FutureResult<Self::Service, Self::InitError>;
fn new_service(&self, _: &()) -> Self::Future {
ok(Decompress(PhantomData))
}
}
impl<P> Service for Decompress<P>
where
P: Stream<Item = Bytes, Error = PayloadError>,
{
type Request = ServiceRequest<P>;
type Response = ServiceRequest<Decoder<Payload<P>>>;
type Error = Error;
type Future = FutureResult<Self::Response, Self::Error>;
fn poll_ready(&mut self) -> Poll<(), Self::Error> {
Ok(Async::Ready(()))
}
fn call(&mut self, req: ServiceRequest<P>) -> Self::Future {
let (req, payload) = req.into_parts();
let payload = Decoder::from_headers(req.headers(), payload);
ok(ServiceRequest::from_parts(req, Payload::Stream(payload)))
}
}

5
src/middleware/mod.rs
View File

@ -4,6 +4,11 @@ mod compress;
#[cfg(any(feature = "brotli", feature = "flate2"))]
pub use self::compress::Compress;
#[cfg(any(feature = "brotli", feature = "flate2"))]
mod decompress;
#[cfg(any(feature = "brotli", feature = "flate2"))]
pub use self::decompress::Decompress;
pub mod cors;
mod defaultheaders;
pub mod errhandlers;

2
src/resource.rs
View File

@ -507,7 +507,7 @@ impl<P> Service for ResourceService<P> {
if let Some(ref mut default) = self.default {
Either::B(Either::A(default.call(req)))
} else {
let req = req.into_request();
let req = req.into_parts().0;
Either::B(Either::B(ok(ServiceResponse::new(
req,
Response::MethodNotAllowed().finish(),

2
src/scope.rs
View File

@ -489,7 +489,7 @@ impl<P> Service for ScopeService<P> {
} else if let Some(ref mut default) = self.default {
Either::A(default.call(req))
} else {
let req = req.into_request();
let req = req.into_parts().0;
Either::B(ok(ServiceResponse::new(req, Response::NotFound().finish())))
}
}

16
src/service.rs
View File

@ -69,9 +69,14 @@ impl<P> ServiceRequest<P> {
}
}
#[inline]
pub fn into_request(self) -> HttpRequest {
self.req
/// Construct service request from parts
pub fn from_parts(req: HttpRequest, payload: Payload<P>) -> Self {
ServiceRequest { req, payload }
}
/// Deconstruct request into parts
pub fn into_parts(self) -> (HttpRequest, Payload<P>) {
(self.req, self.payload)
}
/// Create service response
@ -162,11 +167,6 @@ impl<P> ServiceRequest<P> {
pub fn app_config(&self) -> &AppConfig {
self.req.config()
}
/// Deconstruct request into parts
pub fn into_parts(self) -> (HttpRequest, Payload<P>) {
(self.req, self.payload)
}
}
impl<P> Resource<Url> for ServiceRequest<P> {

3
src/test.rs
View File

@ -350,7 +350,8 @@ impl TestRequest {
Rc::new(self.rmap),
AppConfig::new(self.config),
)
.into_request()
.into_parts()
.0
}
/// Complete request creation and generate `ServiceFromRequest` instance