Support SendResponse::send_continue

src/proto/streams/send.rs

@@ -138,10 +138,11 @@ impl Send {
 
         Self::check_headers(frame.fields())?;
 
+        let final_response = !frame.is_informational();
         let end_stream = frame.is_end_stream();
 
         // Update the state
-        stream.state.send_open(end_stream)?;
+        stream.state.send_open(final_response, end_stream)?;
 
         if counts.peer().is_local_init(frame.stream_id()) {
             // If we're waiting on a PushPromise anyway

src/proto/streams/state.rs

@@ -87,8 +87,12 @@ enum Cause {
 
 impl State {
     /// Opens the send-half of a stream if it is not already open.
-    pub fn send_open(&mut self, eos: bool) -> Result<(), UserError> {
-        let local = Streaming;
+    pub fn send_open(&mut self, final_response: bool, eos: bool) -> Result<(), UserError> {
+        let local = if final_response {
+            Streaming
+        } else {
+            AwaitingHeaders
+        };
 
         self.inner = match self.inner {
             Idle => {
@@ -111,7 +115,7 @@ impl State {
                     Open { local, remote }
                 }
             }
-            HalfClosedRemote(AwaitingHeaders) | ReservedLocal => {
+            HalfClosedRemote(AwaitingHeaders | Streaming) | ReservedLocal => {
                 if eos {
                     Closed(Cause::EndStream)
                 } else {

src/server.rs

@@ -1049,6 +1049,22 @@ impl<B: Buf> SendResponse<B> {
             .map_err(Into::into)
     }
 
+    /// Send a non-final 1xx response to a client request.
+    ///
+    /// The [`SendResponse`] instance is already associated with a received
+    /// request.  This function may only be called if [`send_reset`] or
+    /// [`send_response`] has not been previously called.
+    ///
+    /// [`SendResponse`]: #
+    /// [`send_reset`]: #method.send_reset
+    /// [`send_response`]: #method.send_response
+    pub fn send_info(&mut self, response: Response<()>) -> Result<(), crate::Error> {
+        assert!(response.status().is_informational());
+        self.inner
+            .send_response(response, false)
+            .map_err(Into::into)
+    }
+
     /// Push a request and response to the client
     ///
     /// On success, a [`SendResponse`] instance is returned.

tests/h2-tests/tests/server.rs

@@ -105,6 +105,61 @@ async fn serve_request() {
     join(client, srv).await;
 }
 
+#[tokio::test]
+async fn serve_request_expect_continue() {
+    h2_support::trace_init!();
+    let (io, mut client) = mock::new();
+
+    let client = async move {
+        let settings = client.assert_server_handshake().await;
+        assert_default_settings!(settings);
+        client
+            .send_frame(
+                frames::headers(1)
+                    .field(http::header::EXPECT, "100-continue")
+                    .request("POST", "https://example.com/"),
+            )
+            .await;
+        client.recv_frame(frames::headers(1).response(100)).await;
+        client
+            .send_frame(frames::data(1, "hello world").eos())
+            .await;
+        client
+            .recv_frame(frames::headers(1).response(200).eos())
+            .await;
+    };
+
+    let srv = async move {
+        let mut srv = server::handshake(io).await.expect("handshake");
+        let (req, mut stream) = srv.next().await.unwrap().unwrap();
+
+        assert_eq!(req.method(), &http::Method::POST);
+        assert_eq!(
+            req.headers().get(http::header::EXPECT),
+            Some(&http::HeaderValue::from_static("100-continue"))
+        );
+
+        let connection_fut = poll_fn(|cx| srv.poll_closed(cx).map(Result::ok));
+        let test_fut = async move {
+            stream.send_continue().unwrap();
+
+            let mut body = req.into_body();
+            assert_eq!(
+                body.next().await.unwrap().unwrap(),
+                Bytes::from_static(b"hello world")
+            );
+            assert!(body.next().await.is_none());
+
+            let rsp = http::Response::builder().status(200).body(()).unwrap();
+            stream.send_response(rsp, true).unwrap();
+        };
+
+        join(connection_fut, test_fut).await;
+    };
+
+    join(client, srv).await;
+}
+
 #[tokio::test]
 async fn serve_connect() {
     h2_support::trace_init!();