forked/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(body): replace Chunk type with Bytes

Browse Source 
Closes #1931

BREAKING CHANGE: All usage of `hyper::Chunk` should be replaced with
  `bytes::Bytes` (or `hyper::body::Bytes`).
This commit is contained in:
Sean McArthur
2019-12-05 16:56:35 -08:00
parent c56ccfb033
commit 5a59875742
10 changed files with 59 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/proto/h1/conn.rs
View File

@@ -12,7 +12,6 @@ use super::{/*Decode,*/ Decoder, Encode, EncodedBuf, Encoder, Http1Transaction,
use crate::common::{task, Pin, Poll, Unpin};
use crate::headers::connection_keep_alive;
use crate::proto::{BodyLength, DecodedLength, MessageHead};
use crate::Chunk;
const H2_PREFACE: &'static [u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
@@ -205,7 +204,7 @@ where
pub fn poll_read_body(
&mut self,
cx: &mut task::Context<'_>,
) -> Poll<Option<io::Result<Chunk>>> {
) -> Poll<Option<io::Result<Bytes>>> {
debug_assert!(self.can_read_body());
let (reading, ret) = match self.state.reading {
@@ -217,7 +216,7 @@ where
(
Reading::KeepAlive,
if !slice.is_empty() {
Some(Ok(Chunk::from(slice)))
Some(Ok(slice))
} else {
None
},
@@ -229,7 +228,7 @@ where
// an empty slice...
(Reading::Closed, None)
} else {
return Poll::Ready(Some(Ok(Chunk::from(slice))));
return Poll::Ready(Some(Ok(slice)));
};
(reading, Poll::Ready(chunk))
}
@@ -930,7 +929,7 @@ mod tests {
// an empty IO, we'll be skipping and using the read buffer anyways
let io = tokio_test::io::Builder::new().build();
let mut conn = Conn::<_, crate::Chunk, crate::proto::h1::ServerTransaction>::new(io);
let mut conn = Conn::<_, bytes::Bytes, crate::proto::h1::ServerTransaction>::new(io);
*conn.io.read_buf_mut() = ::bytes::BytesMut::from(&s[..]);
conn.state.cached_headers = Some(HeaderMap::with_capacity(2));
@@ -981,7 +980,7 @@ mod tests {
let good_message = b"GET / HTTP/1.1\r\n\r\n".to_vec();
let len = good_message.len();
let io = AsyncIo::new_buf(good_message, len);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
match conn.poll().unwrap() {
Async::Ready(Some(Frame::Message { message, body: false })) => {
@@ -999,7 +998,7 @@ mod tests {
let _: Result<(), ()> = future::lazy(|| {
let good_message = b"GET / HTTP/1.1\r\nHost: foo.bar\r\n\r\n".to_vec();
let io = AsyncIo::new_buf(good_message, 10);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
assert!(conn.poll().unwrap().is_not_ready());
conn.io.io_mut().block_in(50);
let async = conn.poll().unwrap();
@@ -1015,7 +1014,7 @@ mod tests {
#[test]
fn test_conn_init_read_eof_idle() {
let io = AsyncIo::new_buf(vec![], 1);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.idle();
match conn.poll().unwrap() {
@@ -1027,7 +1026,7 @@ mod tests {
#[test]
fn test_conn_init_read_eof_idle_partial_parse() {
let io = AsyncIo::new_buf(b"GET / HTTP/1.1".to_vec(), 100);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.idle();
match conn.poll() {
@@ -1041,7 +1040,7 @@ mod tests {
let _: Result<(), ()> = future::lazy(|| {
// server ignores
let io = AsyncIo::new_eof();
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.busy();
match conn.poll().unwrap() {
@@ -1051,7 +1050,7 @@ mod tests {
// client
let io = AsyncIo::new_eof();
let mut conn = Conn::<_, proto::Chunk, ClientTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ClientTransaction>::new(io);
conn.state.busy();
match conn.poll() {
@@ -1066,7 +1065,7 @@ mod tests {
fn test_conn_body_finish_read_eof() {
let _: Result<(), ()> = future::lazy(|| {
let io = AsyncIo::new_eof();
let mut conn = Conn::<_, proto::Chunk, ClientTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ClientTransaction>::new(io);
conn.state.busy();
conn.state.writing = Writing::KeepAlive;
conn.state.reading = Reading::Body(Decoder::length(0));
@@ -1091,7 +1090,7 @@ mod tests {
fn test_conn_message_empty_body_read_eof() {
let _: Result<(), ()> = future::lazy(|| {
let io = AsyncIo::new_buf(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n".to_vec(), 1024);
let mut conn = Conn::<_, proto::Chunk, ClientTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ClientTransaction>::new(io);
conn.state.busy();
conn.state.writing = Writing::KeepAlive;
@@ -1115,7 +1114,7 @@ mod tests {
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);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.busy();
match conn.poll() {
@@ -1145,7 +1144,7 @@ mod tests {
#[test]
fn test_conn_closed_read() {
let io = AsyncIo::new_buf(vec![], 0);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.close();
match conn.poll().unwrap() {
@@ -1159,7 +1158,7 @@ mod tests {
let _ = pretty_env_logger::try_init();
let _: Result<(), ()> = future::lazy(|| {
let io = AsyncIo::new_buf(vec![], 0);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
let max = super::super::io::DEFAULT_MAX_BUFFER_SIZE + 4096;
conn.state.writing = Writing::Body(Encoder::length((max * 2) as u64));
@@ -1184,7 +1183,7 @@ mod tests {
fn test_conn_body_write_chunked() {
let _: Result<(), ()> = future::lazy(|| {
let io = AsyncIo::new_buf(vec![], 4096);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.writing = Writing::Body(Encoder::chunked());
assert!(conn.start_send(Frame::Body { chunk: Some("headers".into()) }).unwrap().is_ready());
@@ -1197,7 +1196,7 @@ mod tests {
fn test_conn_body_flush() {
let _: Result<(), ()> = future::lazy(|| {
let io = AsyncIo::new_buf(vec![], 1024 * 1024 * 5);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.writing = Writing::Body(Encoder::length(1024 * 1024));
assert!(conn.start_send(Frame::Body { chunk: Some(vec![b'a'; 1024 * 1024].into()) }).unwrap().is_ready());
assert!(!conn.can_buffer_body());
@@ -1234,7 +1233,7 @@ mod tests {
// test that once writing is done, unparks
let f = future::lazy(|| {
let io = AsyncIo::new_buf(vec![], 4096);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.reading = Reading::KeepAlive;
assert!(conn.poll().unwrap().is_not_ready());
@@ -1248,7 +1247,7 @@ mod tests {
// test that flushing when not waiting on read doesn't unpark
let f = future::lazy(|| {
let io = AsyncIo::new_buf(vec![], 4096);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.writing = Writing::KeepAlive;
assert!(conn.poll_complete().unwrap().is_ready());
Ok::<(), ()>(())
@@ -1259,7 +1258,7 @@ mod tests {
// test that flushing and writing isn't done doesn't unpark
let f = future::lazy(|| {
let io = AsyncIo::new_buf(vec![], 4096);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.reading = Reading::KeepAlive;
assert!(conn.poll().unwrap().is_not_ready());
conn.state.writing = Writing::Body(Encoder::length(5_000));
@@ -1272,7 +1271,7 @@ mod tests {
#[test]
fn test_conn_closed_write() {
let io = AsyncIo::new_buf(vec![], 0);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.close();
match conn.start_send(Frame::Body { chunk: Some(b"foobar".to_vec().into()) }) {
@@ -1286,7 +1285,7 @@ mod tests {
#[test]
fn test_conn_write_empty_chunk() {
let io = AsyncIo::new_buf(vec![], 0);
let mut conn = Conn::<_, proto::Chunk, ServerTransaction>::new(io);
let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io);
conn.state.writing = Writing::KeepAlive;
assert!(conn.start_send(Frame::Body { chunk: None }).unwrap().is_ready());

4
src/proto/h1/dispatch.rs
View File

@@ -646,7 +646,7 @@ mod tests {
// the request is ready to write later...
//let io = AsyncIo::new_buf(b"HTTP/1.1 200 OK\r\n\r\n".to_vec(), 0);
let (mut tx, rx) = crate::client::dispatch::channel();
let conn = Conn::<_, crate::Chunk, ClientTransaction>::new(io);
let conn = Conn::<_, bytes::Bytes, ClientTransaction>::new(io);
let mut dispatcher = Dispatcher::new(Client::new(rx), conn);
// First poll is needed to allow tx to send...
@@ -681,7 +681,7 @@ mod tests {
.build();
let (mut tx, rx) = crate::client::dispatch::channel();
let conn = Conn::<_, crate::Chunk, ClientTransaction>::new(io);
let conn = Conn::<_, bytes::Bytes, ClientTransaction>::new(io);
let mut dispatcher = tokio_test::task::spawn(Dispatcher::new(Client::new(rx), conn));
// First poll is needed to allow tx to send...

4
src/proto/h1/io.rs
View File

@@ -962,10 +962,10 @@ mod tests {
let s = "Hello, World!";
b.bytes = s.len() as u64;
let mut write_buf = WriteBuf::<crate::Chunk>::new();
let mut write_buf = WriteBuf::<bytes::Bytes>::new();
write_buf.set_strategy(WriteStrategy::Flatten);
b.iter(|| {
let chunk = crate::Chunk::from(s);
let chunk = bytes::Bytes::from(s);
write_buf.buffer(chunk);
::test::black_box(&write_buf);
write_buf.headers.bytes.clear();