mirror of
https://github.com/fafhrd91/actix-web
synced 2025-08-31 08:57:00 +02:00
refactor content encoding
This commit is contained in:
Nikolay Kim
266
src/payload.rs
266
src/payload.rs
@@ -1,19 +1,15 @@
|
||||
use std::{io, fmt, cmp};
|
||||
use std::{fmt, cmp};
|
||||
use std::rc::{Rc, Weak};
|
||||
use std::cell::RefCell;
|
||||
use std::collections::VecDeque;
|
||||
use std::error::Error;
|
||||
use std::io::{Read, Write, Error as IoError};
|
||||
use bytes::{Bytes, BytesMut, BufMut, Writer};
|
||||
use std::io::{Error as IoError};
|
||||
use bytes::{Bytes, BytesMut};
|
||||
use http2::Error as Http2Error;
|
||||
use futures::{Async, Poll, Stream};
|
||||
use futures::task::{Task, current as current_task};
|
||||
use flate2::read::{GzDecoder};
|
||||
use flate2::write::{DeflateDecoder};
|
||||
use brotli2::write::BrotliDecoder;
|
||||
|
||||
use actix::ResponseType;
|
||||
use httpresponse::ContentEncoding;
|
||||
|
||||
pub(crate) const DEFAULT_BUFFER_SIZE: usize = 65_536; // max buffer size 64k
|
||||
|
||||
@@ -205,262 +201,6 @@ impl PayloadWriter for PayloadSender {
|
||||
}
|
||||
}
|
||||
|
||||
enum Decoder {
|
||||
Zlib(DeflateDecoder<BytesWriter>),
|
||||
Gzip(Option<GzDecoder<Wrapper>>),
|
||||
Br(Rc<RefCell<BytesMut>>, BrotliDecoder<WrapperRc>),
|
||||
Identity,
|
||||
}
|
||||
|
||||
// should go after write::GzDecoder get implemented
|
||||
#[derive(Debug)]
|
||||
struct Wrapper {
|
||||
buf: BytesMut
|
||||
}
|
||||
|
||||
impl io::Read for Wrapper {
|
||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
let len = cmp::min(buf.len(), self.buf.len());
|
||||
buf[..len].copy_from_slice(&self.buf[..len]);
|
||||
self.buf.split_to(len);
|
||||
Ok(len)
|
||||
}
|
||||
}
|
||||
|
||||
struct BytesWriter {
|
||||
buf: BytesMut,
|
||||
}
|
||||
|
||||
impl Default for BytesWriter {
|
||||
fn default() -> BytesWriter {
|
||||
BytesWriter{buf: BytesMut::with_capacity(8192)}
|
||||
}
|
||||
}
|
||||
|
||||
impl io::Write for BytesWriter {
|
||||
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
||||
self.buf.extend(buf);
|
||||
Ok(buf.len())
|
||||
}
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// should go after brotli2::write::BrotliDecoder::get_mut get implemented
|
||||
#[derive(Debug)]
|
||||
struct WrapperRc {
|
||||
buf: Rc<RefCell<BytesMut>>,
|
||||
}
|
||||
|
||||
impl io::Write for WrapperRc {
|
||||
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
||||
self.buf.borrow_mut().extend(buf);
|
||||
Ok(buf.len())
|
||||
}
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct EncodedPayload {
|
||||
inner: PayloadSender,
|
||||
decoder: Decoder,
|
||||
dst: Writer<BytesMut>,
|
||||
error: bool,
|
||||
}
|
||||
|
||||
impl EncodedPayload {
|
||||
pub fn new(inner: PayloadSender, enc: ContentEncoding) -> EncodedPayload {
|
||||
let dec = match enc {
|
||||
ContentEncoding::Deflate => Decoder::Zlib(
|
||||
DeflateDecoder::new(BytesWriter::default())),
|
||||
ContentEncoding::Gzip => Decoder::Gzip(None),
|
||||
ContentEncoding::Br => {
|
||||
let buf = Rc::new(RefCell::new(BytesMut::new()));
|
||||
let buf2 = Rc::clone(&buf);
|
||||
Decoder::Br(buf, BrotliDecoder::new(WrapperRc{buf: buf2}))
|
||||
}
|
||||
_ => Decoder::Identity,
|
||||
};
|
||||
EncodedPayload {
|
||||
inner: inner,
|
||||
decoder: dec,
|
||||
error: false,
|
||||
dst: BytesMut::new().writer(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl PayloadWriter for EncodedPayload {
|
||||
|
||||
fn set_error(&mut self, err: PayloadError) {
|
||||
self.inner.set_error(err)
|
||||
}
|
||||
|
||||
fn feed_eof(&mut self) {
|
||||
if self.error {
|
||||
return
|
||||
}
|
||||
let err = match self.decoder {
|
||||
Decoder::Br(ref mut buf, ref mut decoder) => {
|
||||
match decoder.flush() {
|
||||
Ok(_) => {
|
||||
let b = buf.borrow_mut().take().freeze();
|
||||
if !b.is_empty() {
|
||||
self.inner.feed_data(b);
|
||||
}
|
||||
self.inner.feed_eof();
|
||||
return
|
||||
},
|
||||
Err(err) => Some(err),
|
||||
}
|
||||
}
|
||||
|
||||
Decoder::Gzip(ref mut decoder) => {
|
||||
if decoder.is_none() {
|
||||
self.inner.feed_eof();
|
||||
return
|
||||
}
|
||||
loop {
|
||||
let len = self.dst.get_ref().len();
|
||||
let len_buf = decoder.as_mut().unwrap().get_mut().buf.len();
|
||||
|
||||
if len < len_buf * 2 {
|
||||
self.dst.get_mut().reserve(len_buf * 2 - len);
|
||||
unsafe{self.dst.get_mut().set_len(len_buf * 2)};
|
||||
}
|
||||
match decoder.as_mut().unwrap().read(&mut self.dst.get_mut()) {
|
||||
Ok(n) => {
|
||||
if n == 0 {
|
||||
self.inner.feed_eof();
|
||||
return
|
||||
} else {
|
||||
self.inner.feed_data(self.dst.get_mut().split_to(n).freeze());
|
||||
}
|
||||
}
|
||||
Err(err) => break Some(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
Decoder::Zlib(ref mut decoder) => {
|
||||
match decoder.flush() {
|
||||
Ok(_) => {
|
||||
let b = decoder.get_mut().buf.take().freeze();
|
||||
if !b.is_empty() {
|
||||
self.inner.feed_data(b);
|
||||
}
|
||||
self.inner.feed_eof();
|
||||
return
|
||||
},
|
||||
Err(err) => Some(err),
|
||||
}
|
||||
},
|
||||
Decoder::Identity => {
|
||||
self.inner.feed_eof();
|
||||
return
|
||||
}
|
||||
};
|
||||
|
||||
self.error = true;
|
||||
self.decoder = Decoder::Identity;
|
||||
if let Some(err) = err {
|
||||
self.set_error(PayloadError::ParseError(err));
|
||||
} else {
|
||||
self.set_error(PayloadError::Incomplete);
|
||||
}
|
||||
}
|
||||
|
||||
fn feed_data(&mut self, data: Bytes) {
|
||||
if self.error {
|
||||
return
|
||||
}
|
||||
match self.decoder {
|
||||
Decoder::Br(ref mut buf, ref mut decoder) => {
|
||||
match decoder.write(&data) {
|
||||
Ok(_) => {
|
||||
let b = buf.borrow_mut().take().freeze();
|
||||
if !b.is_empty() {
|
||||
self.inner.feed_data(b);
|
||||
}
|
||||
return
|
||||
},
|
||||
Err(err) => {
|
||||
trace!("Error decoding br encoding: {}", err);
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
Decoder::Gzip(ref mut decoder) => {
|
||||
if decoder.is_none() {
|
||||
let mut buf = BytesMut::new();
|
||||
buf.extend(data);
|
||||
*decoder = Some(GzDecoder::new(Wrapper{buf: buf}).unwrap());
|
||||
} else {
|
||||
decoder.as_mut().unwrap().get_mut().buf.extend(data);
|
||||
}
|
||||
|
||||
loop {
|
||||
let len_buf = decoder.as_mut().unwrap().get_mut().buf.len();
|
||||
if len_buf == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
let len = self.dst.get_ref().len();
|
||||
if len < len_buf * 2 {
|
||||
self.dst.get_mut().reserve(len_buf * 2 - len);
|
||||
unsafe{self.dst.get_mut().set_len(len_buf * 2)};
|
||||
}
|
||||
match decoder.as_mut().unwrap().read(&mut self.dst.get_mut()) {
|
||||
Ok(n) => {
|
||||
if n == 0 {
|
||||
return
|
||||
} else {
|
||||
self.inner.feed_data(self.dst.get_mut().split_to(n).freeze());
|
||||
}
|
||||
}
|
||||
Err(_) => break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Decoder::Zlib(ref mut decoder) => {
|
||||
match decoder.write(&data) {
|
||||
Ok(_) => {
|
||||
let b = decoder.get_mut().buf.take().freeze();
|
||||
if !b.is_empty() {
|
||||
self.inner.feed_data(b);
|
||||
}
|
||||
return
|
||||
},
|
||||
Err(err) => {
|
||||
trace!("Error decoding deflate encoding: {}", err);
|
||||
},
|
||||
}
|
||||
}
|
||||
Decoder::Identity => {
|
||||
self.inner.feed_data(data);
|
||||
return
|
||||
}
|
||||
};
|
||||
|
||||
self.error = true;
|
||||
self.decoder = Decoder::Identity;
|
||||
self.set_error(PayloadError::EncodingCorrupted);
|
||||
}
|
||||
|
||||
fn capacity(&self) -> usize {
|
||||
match self.decoder {
|
||||
Decoder::Br(ref buf, _) => {
|
||||
buf.borrow().len() + self.inner.capacity()
|
||||
}
|
||||
_ => {
|
||||
self.inner.capacity()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct Inner {
|
||||
|
||||
Reference in New Issue
Block a user