diff --git a/src/proto/h1/conn.rs b/src/proto/h1/conn.rs index 9a968dd0df..7bcad85970 100644 --- a/src/proto/h1/conn.rs +++ b/src/proto/h1/conn.rs @@ -1209,6 +1209,7 @@ mod tests { // (disable keep-alive) at request-encode time, so it is never returned to the // pool for reuse — independent of whether the backend response echoes // `Connection: close`. hyper otherwise derives reuse from the response alone. + #[cfg(feature = "client")] #[test] fn client_request_connection_close_disables_keep_alive() { use super::*; @@ -1264,336 +1265,173 @@ mod tests { ); } - /* - //TODO: rewrite these using dispatch... someday... - use futures::{Async, Future, Stream, Sink}; - use futures::future; - - use proto::{self, ClientTransaction, MessageHead, ServerTransaction}; - use super::super::Encoder; - use mock::AsyncIo; - - use super::{Conn, Decoder, Reading, Writing}; - use ::uri::Uri; - - use std::str::FromStr; - - #[test] - fn test_conn_init_read() { - 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::Bytes, ServerTransaction>::new(io); - - match conn.poll().unwrap() { - Async::Ready(Some(Frame::Message { message, body: false })) => { - assert_eq!(message, MessageHead { - subject: ::proto::RequestLine(::Get, Uri::from_str("/").unwrap()), - .. MessageHead::default() - }) - }, - f => panic!("frame is not Frame::Message: {:?}", f) - } - } - - #[test] - fn test_conn_parse_partial() { - 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::Bytes, ServerTransaction>::new(io); - assert!(conn.poll().unwrap().is_not_ready()); - conn.io.io_mut().block_in(50); - let async = conn.poll().unwrap(); - assert!(async.is_ready()); - match async { - Async::Ready(Some(Frame::Message { .. })) => (), - f => panic!("frame is not Message: {:?}", f), - } - Ok(()) - }).wait(); - } - - #[test] - fn test_conn_init_read_eof_idle() { - let io = AsyncIo::new_buf(vec![], 1); - let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io); - conn.state.idle(); - - match conn.poll().unwrap() { - Async::Ready(None) => {}, - other => panic!("frame is not None: {:?}", other) - } - } - - #[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::Bytes, ServerTransaction>::new(io); - conn.state.idle(); - - match conn.poll() { - Err(ref err) if err.kind() == std::io::ErrorKind::UnexpectedEof => {}, - other => panic!("unexpected frame: {:?}", other) + use super::*; + use crate::common::io::Compat; + #[cfg(feature = "client")] + use crate::proto::h1::ClientTransaction; + #[cfg(feature = "server")] + use crate::proto::h1::ServerTransaction; + #[cfg(feature = "server")] + use crate::proto::RequestLine; + use bytes::Bytes; + + fn poll_head( + conn: &mut Conn, Bytes, T>, + ) -> Poll, DecodedLength, Wants)>>> + where + I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, + T: Http1Transaction + Unpin, + { + tokio_test::task::spawn(()).enter(|cx, _| conn.poll_read_head(cx)) + } + + fn ready(poll: Poll) -> T { + match poll { + Poll::Ready(value) => value, + Poll::Pending => panic!("expected ready"), } } + #[cfg(feature = "server")] #[test] - fn test_conn_init_read_eof_busy() { - let _: Result<(), ()> = future::lazy(|| { - // server ignores - let io = AsyncIo::new_eof(); - let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io); - conn.state.busy(); - - match conn.poll().unwrap() { - Async::Ready(None) => {}, - other => panic!("unexpected frame: {:?}", other) - } + fn conn_reads_request_head() { + let io = tokio_test::io::Builder::new() + .read(b"GET / HTTP/1.1\r\n\r\n") + .build(); + let mut conn = Conn::<_, Bytes, ServerTransaction>::new(Compat::new(io)); - // client - let io = AsyncIo::new_eof(); - let mut conn = Conn::<_, proto::Bytes, ClientTransaction>::new(io); - conn.state.busy(); - - match conn.poll() { - Err(ref err) if err.kind() == std::io::ErrorKind::UnexpectedEof => {}, - other => panic!("unexpected frame: {:?}", other) - } - Ok(()) - }).wait(); + let (head, body, _) = ready(poll_head(&mut conn)) + .expect("message") + .expect("valid request"); + assert_eq!(head.subject, RequestLine(Method::GET, "/".parse().unwrap())); + assert_eq!(body, DecodedLength::ZERO); } + #[cfg(feature = "server")] #[test] - fn test_conn_body_finish_read_eof() { - let _: Result<(), ()> = future::lazy(|| { - let io = AsyncIo::new_eof(); - 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)); - - match conn.poll() { - Ok(Async::Ready(Some(Frame::Body { chunk: None }))) => (), - other => panic!("unexpected frame: {:?}", other) - } - - // conn eofs, but tokio-proto will call poll() again, before calling flush() - // the conn eof in this case is perfectly fine - - match conn.poll() { - Ok(Async::Ready(None)) => (), - other => panic!("unexpected frame: {:?}", other) - } - Ok(()) - }).wait(); + fn conn_reads_partial_request_head() { + tokio_test::task::spawn(()).enter(|cx, _| { + let (io, mut handle) = tokio_test::io::Builder::new().build_with_handle(); + let mut conn = Conn::<_, Bytes, ServerTransaction>::new(Compat::new(io)); + handle.read(b"GET / HTTP"); + assert!(conn.poll_read_head(cx).is_pending()); + handle.read(b"/1.1\r\nHost: foo.bar\r\n\r\n"); + assert!(conn.poll_read_head(cx).is_ready()); + }); } + #[cfg(feature = "server")] #[test] - 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::Bytes, ClientTransaction>::new(io); - conn.state.busy(); - conn.state.writing = Writing::KeepAlive; - - match conn.poll() { - Ok(Async::Ready(Some(Frame::Message { body: false, .. }))) => (), - other => panic!("unexpected frame: {:?}", other) - } - - // conn eofs, but tokio-proto will call poll() again, before calling flush() - // the conn eof in this case is perfectly fine - - match conn.poll() { - Ok(Async::Ready(None)) => (), - other => panic!("unexpected frame: {:?}", other) - } - Ok(()) - }).wait(); + fn conn_accepts_eof_when_idle() { + let io = tokio_test::io::Builder::new().build(); + let mut conn = Conn::<_, Bytes, ServerTransaction>::new(Compat::new(io)); + conn.state.idle::(); + assert!(matches!(poll_head(&mut conn), Poll::Ready(None))); } + #[cfg(feature = "server")] #[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::Bytes, ServerTransaction>::new(io); - 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); - let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io); - conn.state.close(); - - match conn.poll().unwrap() { - Async::Ready(None) => {}, - other => panic!("frame is not None: {:?}", other) - } + fn conn_rejects_eof_during_partial_head() { + let io = tokio_test::io::Builder::new() + .read(b"GET / HTTP/1.1") + .build(); + let mut conn = Conn::<_, Bytes, ServerTransaction>::new(Compat::new(io)); + conn.state.idle::(); + let err = ready(poll_head(&mut conn)) + .expect("error result") + .expect_err("partial head must fail"); + assert!(err.is_incomplete_message(), "unexpected error: {err:?}"); } + #[cfg(feature = "client")] #[test] - fn test_conn_body_write_length() { - let _ = pretty_env_logger::try_init(); - let _: Result<(), ()> = future::lazy(|| { - let io = AsyncIo::new_buf(vec![], 0); - 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)); - - assert!(conn.start_send(Frame::Body { chunk: Some(vec![b'a'; max].into()) }).unwrap().is_ready()); - assert!(!conn.can_buffer_body()); - - assert!(conn.start_send(Frame::Body { chunk: Some(vec![b'b'; 1024 * 8].into()) }).unwrap().is_not_ready()); - - conn.io.io_mut().block_in(1024 * 3); - assert!(conn.poll_complete().unwrap().is_not_ready()); - conn.io.io_mut().block_in(1024 * 3); - assert!(conn.poll_complete().unwrap().is_not_ready()); - conn.io.io_mut().block_in(max * 2); - assert!(conn.poll_complete().unwrap().is_ready()); - - assert!(conn.start_send(Frame::Body { chunk: Some(vec![b'c'; 1024 * 8].into()) }).unwrap().is_ready()); - Ok(()) - }).wait(); + fn client_rejects_eof_while_busy() { + let io = tokio_test::io::Builder::new().build(); + let mut client = Conn::<_, Bytes, ClientTransaction>::new(Compat::new(io)); + client.state.busy(); + client.state.writing = Writing::KeepAlive; + let err = ready(poll_head(&mut client)) + .expect("error result") + .expect_err("client EOF must fail"); + assert!(err.is_incomplete_message(), "unexpected error: {err:?}"); } + #[cfg(feature = "server")] #[test] - fn test_conn_body_write_chunked() { - let _: Result<(), ()> = future::lazy(|| { - let io = AsyncIo::new_buf(vec![], 4096); - 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()); - assert!(conn.start_send(Frame::Body { chunk: Some(vec![b'x'; 8192].into()) }).unwrap().is_ready()); - Ok(()) - }).wait(); + fn server_accepts_eof_while_busy() { + let io = tokio_test::io::Builder::new().build(); + let mut server = Conn::<_, Bytes, ServerTransaction>::new(Compat::new(io)); + server.state.busy(); + assert!(matches!(poll_head(&mut server), Poll::Ready(None))); } + #[cfg(feature = "client")] #[test] - fn test_conn_body_flush() { - let _: Result<(), ()> = future::lazy(|| { - let io = AsyncIo::new_buf(vec![], 1024 * 1024 * 5); - 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()); - conn.io.io_mut().block_in(1024 * 1024 * 5); - assert!(conn.poll_complete().unwrap().is_ready()); - assert!(conn.can_buffer_body()); - assert!(conn.io.io_mut().flushed()); - - Ok(()) - }).wait(); + fn conn_reads_empty_response_before_eof() { + let io = tokio_test::io::Builder::new() + .read(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") + .build(); + let mut conn = Conn::<_, Bytes, ClientTransaction>::new(Compat::new(io)); + conn.state.busy(); + conn.state.writing = Writing::KeepAlive; + let (_, body, _) = ready(poll_head(&mut conn)) + .expect("response") + .expect("valid response"); + assert_eq!(body, DecodedLength::ZERO); } + #[cfg(feature = "server")] #[test] - fn test_conn_parking() { - use std::sync::Arc; - use futures::executor::Notify; - use futures::executor::NotifyHandle; - - struct Car { - permit: bool, - } - impl Notify for Car { - fn notify(&self, _id: usize) { - assert!(self.permit, "unparked without permit"); - } - } - - fn car(permit: bool) -> NotifyHandle { - Arc::new(Car { - permit: permit, - }).into() - } - - // test that once writing is done, unparks - let f = future::lazy(|| { - let io = AsyncIo::new_buf(vec![], 4096); - 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::KeepAlive; - assert!(conn.poll_complete().unwrap().is_ready()); - Ok::<(), ()>(()) - }); - ::futures::executor::spawn(f).poll_future_notify(&car(true), 0).unwrap(); - - - // 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::Bytes, ServerTransaction>::new(io); - conn.state.writing = Writing::KeepAlive; - assert!(conn.poll_complete().unwrap().is_ready()); - Ok::<(), ()>(()) - }); - ::futures::executor::spawn(f).poll_future_notify(&car(false), 0).unwrap(); - - - // 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::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)); - assert!(conn.poll_complete().unwrap().is_ready()); - Ok::<(), ()>(()) + fn conn_reads_body_and_reports_end() { + let io = tokio_test::io::Builder::new() + .read(b"POST / HTTP/1.1\r\nContent-Length: 5\r\n\r\n12345") + .wait(std::time::Duration::from_secs(1)) + .build(); + let mut conn = Conn::<_, Bytes, ServerTransaction>::new(Compat::new(io)); + let (_, body, _) = ready(poll_head(&mut conn)) + .expect("request") + .expect("valid request"); + assert_eq!(body, DecodedLength::new(5)); + + tokio_test::task::spawn(()).enter(|cx, _| { + let frame = conn.poll_read_body(cx); + let data = ready(frame) + .expect("body frame") + .expect("valid body") + .into_data() + .expect("data frame"); + assert_eq!(data, "12345"); + assert!( + !conn.can_read_body(), + "the complete body must return to head-reading state" + ); }); - ::futures::executor::spawn(f).poll_future_notify(&car(false), 0).unwrap(); } + #[cfg(feature = "server")] #[test] - fn test_conn_closed_write() { - let io = AsyncIo::new_buf(vec![], 0); - let mut conn = Conn::<_, proto::Bytes, ServerTransaction>::new(io); + fn closed_conn_cannot_read_or_write() { + let io = tokio_test::io::Builder::new().build(); + let mut conn = Conn::<_, Bytes, ServerTransaction>::new(Compat::new(io)); conn.state.close(); - - match conn.start_send(Frame::Body { chunk: Some(b"foobar".to_vec().into()) }) { - Err(_e) => {}, - other => panic!("did not return Err: {:?}", other) - } - - assert!(conn.state.is_write_closed()); + assert!(conn.is_read_closed()); + assert!(conn.is_write_closed()); + assert!(!conn.can_read_head()); + assert!(!conn.can_write_head()); } + #[cfg(feature = "server")] #[test] - fn test_conn_write_empty_chunk() { - let io = AsyncIo::new_buf(vec![], 0); - 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()); - assert!(conn.start_send(Frame::Body { chunk: Some(Vec::new().into()) }).unwrap().is_ready()); - conn.start_send(Frame::Body { chunk: Some(vec![b'a'].into()) }).unwrap_err(); + fn conn_writes_chunked_body() { + let io = tokio_test::io::Builder::new() + .write(b"7\r\nheaders\r\n0\r\n\r\n") + .build(); + let mut conn = Conn::<_, Bytes, ServerTransaction>::new(Compat::new(io)); + conn.state.writing = Writing::Body(Encoder::chunked()); + conn.write_body(Bytes::from_static(b"headers")); + conn.end_body().unwrap(); + tokio_test::task::spawn(()).enter(|cx, _| { + assert!(conn.poll_flush(cx).is_ready()); + }); } - */ }