diff --git a/src/proto/h1/dispatch.rs b/src/proto/h1/dispatch.rs index 6ccbecba94..4dd405d185 100644 --- a/src/proto/h1/dispatch.rs +++ b/src/proto/h1/dispatch.rs @@ -169,6 +169,9 @@ where ready!(self.poll_loop(cx))?; if self.is_done() { + if self.conn.has_buffered_write() { + ready!(self.poll_flush(cx))?; + } if let Some(pending) = self.conn.pending_upgrade() { self.conn.take_error()?; return Poll::Ready(Ok(Dispatched::Upgrade(pending))); diff --git a/tests/server.rs b/tests/server.rs index 094558cb61..54ba903b00 100644 --- a/tests/server.rs +++ b/tests/server.rs @@ -1753,6 +1753,55 @@ async fn upgrades() { assert_eq!(vec, b"bar=foo"); } +#[tokio::test] +async fn upgrades_flush_buffered_response_before_handoff() { + let prefix = b"HTTP/1.1"; + let (mut client, server) = tokio::io::duplex(prefix.len()); + + let conn = http1::Builder::new().serve_connection( + TokioIo::new(server), + service_fn(|_| { + let res = Response::builder() + .status(101) + .header("upgrade", "foobar") + .body(Empty::::new()) + .unwrap(); + future::ready(Ok::<_, hyper::Error>(res)) + }), + ); + let child = tokio::spawn(async move { + let parts = conn.without_shutdown().await.unwrap(); + let mut io = parts.io.inner(); + io.write_all(b"foo=bar").await.unwrap(); + io.shutdown().await.unwrap(); + }); + + client + .write_all( + b"\ + POST / HTTP/1.1\r\n\ + Upgrade: foobar\r\n\ + Connection: upgrade\r\n\ + Content-Length: 5\r\n\ + \r\n\ + ", + ) + .await + .unwrap(); + let mut head = vec![0; prefix.len()]; + client.read_exact(&mut head).await.unwrap(); + assert_eq!(head, prefix); + client.write_all(b"hello").await.unwrap(); + let mut buf = head; + client.read_to_end(&mut buf).await.unwrap(); + child.await.unwrap(); + + let got = s(&buf); + let expected = "HTTP/1.1 101 Switching Protocols\r\n"; + assert!(got.starts_with(expected), "{:?}", got); + assert!(got.ends_with("\r\n\r\nfoo=bar"), "{:?}", got); +} + #[tokio::test] async fn http_connect() { let (listener, addr) = setup_tcp_listener();