Skip to content

Commit

Permalink
fix(tokio-proto): return end-of-body frame correctly for tokio-proto
Browse files Browse the repository at this point in the history
Closes #1414
  • Loading branch information
seanmonstar committed Jan 16, 2018
1 parent 73511ac commit 14e4c74
Show file tree
Hide file tree
Showing 2 changed files with 40 additions and 14 deletions.
50 changes: 39 additions & 11 deletions src/proto/conn.rs
Original file line number Diff line number Diff line change
Expand Up @@ -237,20 +237,17 @@ where I: AsyncRead + AsyncWrite,
Reading::Body(ref mut decoder) => {
match decoder.decode(&mut self.io) {
Ok(Async::Ready(slice)) => {
let chunk = if !slice.is_empty() {
Some(super::Chunk::from(slice))
} else {
None
};
let reading = if decoder.is_eof() {
let (reading, chunk) = if !slice.is_empty() {
return Ok(Async::Ready(Some(super::Chunk::from(slice))));
} else if decoder.is_eof() {
debug!("incoming body completed");
Reading::KeepAlive
} else if chunk.is_some() {
Reading::Body(decoder.clone())
(Reading::KeepAlive, None)
} else {
trace!("decode stream unexpectedly ended");
//TODO: Should this return an UnexpectedEof?
Reading::Closed
// this should actually be unreachable:
// the decoder will return an UnexpectedEof if there were
// no bytes to read and it isn't eof yet...
(Reading::Closed, None)
};
(reading, Ok(Async::Ready(chunk)))
},
Expand Down Expand Up @@ -1078,6 +1075,37 @@ mod tests {
}).wait();
}

#[test]
fn test_conn_read_body_end() {
let _: Result<(), ()> = future::lazy(|| {
let io = AsyncIo::new_buf(b"POST / HTTP/1.1\r\nContent-Length: 5\r\n\r\n12345".to_vec(), 1024);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io, Default::default());
conn.state.busy();

match conn.poll() {
Ok(Async::Ready(Some(Frame::Message { body: true, .. }))) => (),
other => panic!("unexpected frame: {:?}", other)
}

match conn.poll() {
Ok(Async::Ready(Some(Frame::Body { chunk: Some(_) }))) => (),
other => panic!("unexpected frame: {:?}", other)
}

// When the body is done, `poll` MUST return a `Body` frame with chunk set to `None`
match conn.poll() {
Ok(Async::Ready(Some(Frame::Body { chunk: None }))) => (),
other => panic!("unexpected frame: {:?}", other)
}

match conn.poll() {
Ok(Async::NotReady) => (),
other => panic!("unexpected frame: {:?}", other)
}
Ok(())
}).wait();
}

#[test]
fn test_conn_closed_read() {
let io = AsyncIo::new_buf(vec![], 0);
Expand Down
4 changes: 1 addition & 3 deletions src/proto/h1/decode.rs
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,6 @@ impl Decoder {
// methods

pub fn is_eof(&self) -> bool {
trace!("is_eof? {:?}", self);
match self.kind {
Length(0) |
Chunked(ChunkedState::End, _) |
Expand All @@ -85,16 +84,15 @@ impl Decoder {
}

pub fn decode<R: MemRead>(&mut self, body: &mut R) -> Poll<Bytes, io::Error> {
trace!("decode; state={:?}", self.kind);
match self.kind {
Length(ref mut remaining) => {
trace!("Sized read, remaining={:?}", remaining);
if *remaining == 0 {
Ok(Async::Ready(Bytes::new()))
} else {
let to_read = *remaining as usize;
let buf = try_ready!(body.read_mem(to_read));
let num = buf.as_ref().len() as u64;
trace!("Length read: {}", num);
if num > *remaining {
*remaining = 0;
} else if num == 0 {
Expand Down

0 comments on commit 14e4c74

Please sign in to comment.