actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2025-07-01 16:55:08 +02:00
Code Issues Releases Wiki Activity

impl response body support

Browse Source
This commit is contained in:
Nikolay Kim
2018-10-08 15:24:51 -07:00
parent 431e33acb2
commit 805e7a4cd0
4 changed files with 427 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/h1/codec.rs
View File

@ -6,7 +6,7 @@ use tokio_codec::{Decoder, Encoder};
use super::decoder::{PayloadDecoder, PayloadItem, RequestDecoder};
use super::encoder::{ResponseEncoder, ResponseLength};
use body::Body;
use body::{Binary, Body};
use config::ServiceConfig;
use error::ParseError;
use helpers;
@ -26,12 +26,13 @@ bitflags! {
const AVERAGE_HEADER_SIZE: usize = 30;
#[derive(Debug)]
/// Http response
pub enum OutMessage {
/// Http response message
Response(Response),
/// Payload chunk
Payload(Bytes),
Payload(Option<Binary>),
}
/// Incoming http/1 request
@ -151,6 +152,7 @@ impl Codec {
buffer.extend_from_slice(reason);
// content length
let mut len_is_set = true;
match self.te.length {
ResponseLength::Chunked => {
buffer.extend_from_slice(b"\r\ntransfer-encoding: chunked\r\n")
@ -167,6 +169,10 @@ impl Codec {
buffer.extend_from_slice(b"\r\n");
}
ResponseLength::None => buffer.extend_from_slice(b"\r\n"),
ResponseLength::HeaderOrZero => {
len_is_set = false;
buffer.extend_from_slice(b"\r\n")
}
}
// write headers
@ -179,6 +185,9 @@ impl Codec {
TRANSFER_ENCODING => continue,
CONTENT_LENGTH => match self.te.length {
ResponseLength::None => (),
ResponseLength::HeaderOrZero => {
len_is_set = true;
}
_ => continue,
},
DATE => {
@ -215,11 +224,13 @@ impl Codec {
unsafe {
buffer.advance_mut(pos);
}
if !len_is_set {
buffer.extend_from_slice(b"content-length: 0\r\n")
}
// optimized date header, set_date writes \r\n
if !has_date {
self.config.set_date(buffer);
buffer.extend_from_slice(b"\r\n");
} else {
// msg eof
buffer.extend_from_slice(b"\r\n");
@ -272,8 +283,11 @@ impl Encoder for Codec {
OutMessage::Response(res) => {
self.encode_response(res, dst)?;
}
OutMessage::Payload(bytes) => {
dst.extend_from_slice(&bytes);
OutMessage::Payload(Some(bytes)) => {
self.te.encode(bytes.as_ref(), dst)?;
}
OutMessage::Payload(None) => {
self.te.encode_eof(dst)?;
}
}
Ok(())

134
src/h1/dispatcher.rs
View File

@ -12,7 +12,7 @@ use tokio_timer::Delay;
use error::{ParseError, PayloadError};
use payload::{Payload, PayloadSender, PayloadStatus, PayloadWriter};
use body::Body;
use body::{Body, BodyStream};
use config::ServiceConfig;
use error::DispatchError;
use request::Request;
@ -61,9 +61,8 @@ enum Message {
enum State<S: Service> {
None,
ServiceCall(S::Future),
SendResponse(Option<OutMessage>),
SendResponseWithPayload(Option<(OutMessage, Body)>),
Payload(Body),
SendResponse(Option<(OutMessage, Body)>),
SendPayload(Option<BodyStream>, Option<OutMessage>),
}
impl<S: Service> State<S> {
@ -99,6 +98,7 @@ where
};
let framed = Framed::new(stream, Codec::new(config.clone()));
// keep-alive timer
let (ka_expire, ka_timer) = if let Some(delay) = timeout {
(delay.deadline(), Some(delay))
} else if let Some(delay) = config.keep_alive_timer() {
@ -174,59 +174,32 @@ where
break if let Some(msg) = self.messages.pop_front() {
match msg {
Message::Item(req) => Some(self.handle_request(req)?),
Message::Error(res) => Some(State::SendResponse(Some(
Message::Error(res) => Some(State::SendResponse(Some((
OutMessage::Response(res),
))),
Body::Empty,
)))),
}
} else {
None
};
},
State::Payload(ref mut _body) => unimplemented!(),
State::ServiceCall(ref mut fut) => match fut
.poll()
.map_err(DispatchError::Service)?
{
Async::Ready(mut res) => {
self.framed.get_codec_mut().prepare_te(&mut res);
let body = res.replace_body(Body::Empty);
if body.is_empty() {
Some(State::SendResponse(Some(OutMessage::Response(res))))
} else {
Some(State::SendResponseWithPayload(Some((
// call inner service
State::ServiceCall(ref mut fut) => {
match fut.poll().map_err(DispatchError::Service)? {
Async::Ready(mut res) => {
self.framed.get_codec_mut().prepare_te(&mut res);
let body = res.replace_body(Body::Empty);
Some(State::SendResponse(Some((
OutMessage::Response(res),
body,
))))
}
}
Async::NotReady => None,
},
State::SendResponse(ref mut item) => {
let msg = item.take().expect("SendResponse is empty");
match self.framed.start_send(msg) {
Ok(AsyncSink::Ready) => {
self.flags.set(
Flags::KEEPALIVE,
self.framed.get_codec().keepalive(),
);
self.flags.remove(Flags::FLUSHED);
Some(State::None)
}
Ok(AsyncSink::NotReady(msg)) => {
*item = Some(msg);
return Ok(());
}
Err(err) => {
if let Some(mut payload) = self.payload.take() {
payload.set_error(PayloadError::Incomplete);
}
return Err(DispatchError::Io(err));
}
Async::NotReady => None,
}
}
State::SendResponseWithPayload(ref mut item) => {
let (msg, body) =
item.take().expect("SendResponseWithPayload is empty");
// send respons
State::SendResponse(ref mut item) => {
let (msg, body) = item.take().expect("SendResponse is empty");
match self.framed.start_send(msg) {
Ok(AsyncSink::Ready) => {
self.flags.set(
@ -234,7 +207,16 @@ where
self.framed.get_codec().keepalive(),
);
self.flags.remove(Flags::FLUSHED);
Some(State::Payload(body))
match body {
Body::Empty => Some(State::None),
Body::Binary(bin) => Some(State::SendPayload(
None,
Some(OutMessage::Payload(bin.into())),
)),
Body::Streaming(stream) => {
Some(State::SendPayload(Some(stream), None))
}
}
}
Ok(AsyncSink::NotReady(msg)) => {
*item = Some((msg, body));
@ -248,6 +230,48 @@ where
}
}
}
// Send payload
State::SendPayload(ref mut stream, ref mut bin) => {
if let Some(item) = bin.take() {
match self.framed.start_send(item) {
Ok(AsyncSink::Ready) => {
self.flags.remove(Flags::FLUSHED);
}
Ok(AsyncSink::NotReady(item)) => {
*bin = Some(item);
return Ok(());
}
Err(err) => return Err(DispatchError::Io(err)),
}
}
if let Some(ref mut stream) = stream {
match stream.poll() {
Ok(Async::Ready(Some(item))) => match self
.framed
.start_send(OutMessage::Payload(Some(item.into())))
{
Ok(AsyncSink::Ready) => {
self.flags.remove(Flags::FLUSHED);
continue;
}
Ok(AsyncSink::NotReady(msg)) => {
*bin = Some(msg);
return Ok(());
}
Err(err) => return Err(DispatchError::Io(err)),
},
Ok(Async::Ready(None)) => Some(State::SendPayload(
None,
Some(OutMessage::Payload(None)),
)),
Ok(Async::NotReady) => return Ok(()),
// Err(err) => return Err(DispatchError::Io(err)),
Err(_) => return Err(DispatchError::Unknown),
}
} else {
Some(State::None)
}
}
};
match state {
@ -275,19 +299,13 @@ where
Async::Ready(mut res) => {
self.framed.get_codec_mut().prepare_te(&mut res);
let body = res.replace_body(Body::Empty);
if body.is_empty() {
Ok(State::SendResponse(Some(OutMessage::Response(res))))
} else {
Ok(State::SendResponseWithPayload(Some((
OutMessage::Response(res),
body,
))))
}
Ok(State::SendResponse(Some((OutMessage::Response(res), body))))
}
Async::NotReady => Ok(State::ServiceCall(task)),
}
}
/// Process one incoming message
fn one_message(&mut self, msg: InMessage) -> Result<(), DispatchError<S::Error>> {
self.flags.insert(Flags::STARTED);
@ -408,10 +426,12 @@ where
// timeout on first request (slow request) return 408
trace!("Slow request timeout");
self.flags.insert(Flags::STARTED | Flags::DISCONNECTED);
self.state =
State::SendResponse(Some(OutMessage::Response(
self.state = State::SendResponse(Some((
OutMessage::Response(
Response::RequestTimeout().finish(),
)));
),
Body::Empty,
)));
}
} else if let Some(deadline) = self.config.keep_alive_expire() {
timer.reset(deadline)

46
src/h1/encoder.rs
View File

@ -17,9 +17,13 @@ use response::Response;
#[derive(Debug)]
pub(crate) enum ResponseLength {
Chunked,
/// Content length is 0
Zero,
/// Check if headers contains length or write 0
HeaderOrZero,
Length(usize),
Length64(u64),
/// Do no set content-length
None,
}
@ -41,6 +45,16 @@ impl Default for ResponseEncoder {
}
impl ResponseEncoder {
/// Encode message
pub fn encode(&mut self, msg: &[u8], buf: &mut BytesMut) -> io::Result<bool> {
self.te.encode(msg, buf)
}
/// Encode eof
pub fn encode_eof(&mut self, buf: &mut BytesMut) -> io::Result<()> {
self.te.encode_eof(buf)
}
pub fn update(&mut self, resp: &mut Response, head: bool, version: Version) {
self.head = head;
@ -63,17 +77,13 @@ impl ResponseEncoder {
let transfer = match resp.body() {
Body::Empty => {
if !self.head {
self.length = match resp.status() {
StatusCode::NO_CONTENT
| StatusCode::CONTINUE
| StatusCode::SWITCHING_PROTOCOLS
| StatusCode::PROCESSING => ResponseLength::None,
_ => ResponseLength::Zero,
};
} else {
self.length = ResponseLength::Zero;
}
self.length = match resp.status() {
StatusCode::NO_CONTENT
| StatusCode::CONTINUE
| StatusCode::SWITCHING_PROTOCOLS
| StatusCode::PROCESSING => ResponseLength::None,
_ => ResponseLength::HeaderOrZero,
};
TransferEncoding::empty()
}
Body::Binary(_) => {
@ -253,16 +263,22 @@ impl TransferEncoding {
/// Encode eof. Return `EOF` state of encoder
#[inline]
pub fn encode_eof(&mut self, buf: &mut BytesMut) -> bool {
pub fn encode_eof(&mut self, buf: &mut BytesMut) -> io::Result<()> {
match self.kind {
TransferEncodingKind::Eof => true,
TransferEncodingKind::Length(rem) => rem == 0,
TransferEncodingKind::Eof => Ok(()),
TransferEncodingKind::Length(rem) => {
if rem != 0 {
Err(io::Error::new(io::ErrorKind::UnexpectedEof, ""))
} else {
Ok(())
}
}
TransferEncodingKind::Chunked(ref mut eof) => {
if !*eof {
*eof = true;
buf.extend_from_slice(b"0\r\n\r\n");
}
true
Ok(())
}
}
}