prototype support for 1xx informational responses

This is a prototype intended to spur discussion about what support for 1xx
informational responses should look like in a Hyper server.  The good news is
that it works great (for HTTP/1 only, so far).  The bad news is it's kind of
ugly.  Here's what I did:

- Add `ext::InformationalSender`, a type which wraps a
  `futures_channel::mspc::Sender<Response<()>>`.  This may be added as an
  extension to an inbound `Request` by the Hyper server, and the application
  and/or middleware may use it to send one or more informational responses
  before sending the real one.

- Add code to `proto::h1::dispatch` and friends to add such an extension to each
  inbound request and then poll the `Receiver` end along with the future
  representing the final response.  If the app never sends any informational
  responses, then everything proceeds as normal.  Otherwise, we send those
  responses as they become available until the final response is ready.

TODO hyperium#1: Also support informational responses in the HTTP/2 server.
TODO hyperium#2: Come up with a less hacky API?
TODO hyperium#3: Add test coverage.

Signed-off-by: Joel Dice <[email protected]>

src/error.rs

@@ -143,10 +143,6 @@ pub(super) enum User {
     #[cfg(any(feature = "http1", feature = "http2"))]
     #[cfg(feature = "server")]
     UnexpectedHeader,
-    /// User tried to respond with a 1xx (not 101) response code.
-    #[cfg(feature = "http1")]
-    #[cfg(feature = "server")]
-    UnsupportedStatusCode,
 
     /// User tried polling for an upgrade that doesn't exist.
     NoUpgrade,
@@ -360,12 +356,6 @@ impl Error {
         Error::new(Kind::HeaderTimeout)
     }
 
-    #[cfg(feature = "http1")]
-    #[cfg(feature = "server")]
-    pub(super) fn new_user_unsupported_status_code() -> Error {
-        Error::new_user(User::UnsupportedStatusCode)
-    }
-
     pub(super) fn new_user_no_upgrade() -> Error {
         Error::new_user(User::NoUpgrade)
     }
@@ -496,11 +486,6 @@ impl Error {
             #[cfg(any(feature = "http1", feature = "http2"))]
             #[cfg(feature = "server")]
             Kind::User(User::UnexpectedHeader) => "user sent unexpected header",
-            #[cfg(feature = "http1")]
-            #[cfg(feature = "server")]
-            Kind::User(User::UnsupportedStatusCode) => {
-                "response has 1xx status code, not supported by server"
-            }
             Kind::User(User::NoUpgrade) => "no upgrade available",
             #[cfg(all(any(feature = "client", feature = "server"), feature = "http1"))]
             Kind::User(User::ManualUpgrade) => "upgrade expected but low level API in use",

src/ext/mod.rs

@@ -235,3 +235,13 @@ impl OriginalHeaderOrder {
         self.entry_order.iter()
     }
 }
+
+/// Request extension type for sending one or more 1xx informational responses
+/// prior to the final response.
+///
+/// This extension is meant to be attached to inbound `Request`s, allowing a
+/// server to send informational responses immediately (i.e. without delaying
+/// them until it has constructed a final, non-informational response).
+#[cfg(all(feature = "server", any(feature = "http1", feature = "http2")))]
+#[derive(Clone, Debug)]
+pub struct InformationalSender(pub futures_channel::mpsc::Sender<http::Response<()>>);

src/proto/h1/conn.rs

@@ -642,7 +642,7 @@ where
                         head.extensions.remove::<crate::ffi::OnInformational>();
                 }
 
-                Some(encoder)
+                encoder
             }
             Err(err) => {
                 self.state.error = Some(err);

src/proto/h1/dispatch.rs

@@ -8,14 +8,18 @@ use std::{
 
 use crate::rt::{Read, Write};
 use bytes::{Buf, Bytes};
-use futures_util::ready;
-use http::Request;
+#[cfg(feature = "server")]
+use futures_channel::mpsc::{self, Receiver};
+use futures_util::{ready, StreamExt};
+use http::{Request, Response};
 
 use super::{Http1Transaction, Wants};
 use crate::body::{Body, DecodedLength, Incoming as IncomingBody};
 #[cfg(feature = "client")]
 use crate::client::dispatch::TrySendError;
 use crate::common::task;
+#[cfg(feature = "server")]
+use crate::ext::InformationalSender;
 use crate::proto::{BodyLength, Conn, Dispatched, MessageHead, RequestHead};
 use crate::upgrade::OnUpgrade;
 
@@ -35,7 +39,7 @@ pub(crate) trait Dispatch {
     fn poll_msg(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), Self::PollError>>>;
+    ) -> Poll<Option<Result<(Self::PollItem, Option<Self::PollBody>), Self::PollError>>>;
     fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, IncomingBody)>)
         -> crate::Result<()>;
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), ()>>;
@@ -46,6 +50,7 @@ cfg_server! {
     use crate::service::HttpService;
 
     pub(crate) struct Server<S: HttpService<B>, B> {
+        informational_rx: Option<Receiver<Response<()>>>,
         in_flight: Pin<Box<Option<S::Future>>>,
         pub(crate) service: S,
     }
@@ -336,17 +341,22 @@ where
                 if let Some(msg) = ready!(Pin::new(&mut self.dispatch).poll_msg(cx)) {
                     let (head, body) = msg.map_err(crate::Error::new_user_service)?;
 
-                    let body_type = if body.is_end_stream() {
+                    let body_type = if let Some(body) = body {
+                        if body.is_end_stream() {
+                            self.body_rx.set(None);
+                            None
+                        } else {
+                            let btype = body
+                                .size_hint()
+                                .exact()
+                                .map(BodyLength::Known)
+                                .or(Some(BodyLength::Unknown));
+                            self.body_rx.set(Some(body));
+                            btype
+                        }
+                    } else {
                         self.body_rx.set(None);
                         None
-                    } else {
-                        let btype = body
-                            .size_hint()
-                            .exact()
-                            .map(BodyLength::Known)
-                            .or(Some(BodyLength::Unknown));
-                        self.body_rx.set(Some(body));
-                        btype
                     };
                     self.conn.write_head(head, body_type);
                 } else {
@@ -505,6 +515,7 @@ cfg_server! {
     {
         pub(crate) fn new(service: S) -> Server<S, B> {
             Server {
+                informational_rx: None,
                 in_flight: Box::pin(None),
                 service,
             }
@@ -532,8 +543,32 @@ cfg_server! {
         fn poll_msg(
             mut self: Pin<&mut Self>,
             cx: &mut Context<'_>,
-        ) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), Self::PollError>>> {
+        ) -> Poll<Option<Result<(Self::PollItem, Option<Self::PollBody>), Self::PollError>>> {
             let mut this = self.as_mut();
+
+            if let Some(informational_rx) = this.informational_rx.as_mut() {
+                if let Poll::Ready(informational) = informational_rx.poll_next_unpin(cx) {
+                    if let Some(informational) = informational {
+                        let (parts, _) = informational.into_parts();
+                        if parts.status.is_informational() {
+                            let head = MessageHead {
+                                version: parts.version,
+                                subject: parts.status,
+                                headers: parts.headers,
+                                extensions: parts.extensions,
+                            };
+                            return Poll::Ready(Some(Ok((head, None))));
+                        } else {
+                            // TODO: should return an error here, but we have no way of creating a
+                            // `Self::PollError`; might need to change the signature of `Dispatch::poll_msg`.
+                            return Poll::Ready(None);
+                        }
+                    } else {
+                        this.informational_rx = None;
+                    }
+                }
+            }
+
             let ret = if let Some(ref mut fut) = this.in_flight.as_mut().as_pin_mut() {
                 let resp = ready!(fut.as_mut().poll(cx)?);
                 let (parts, body) = resp.into_parts();
@@ -543,13 +578,14 @@ cfg_server! {
                     headers: parts.headers,
                     extensions: parts.extensions,
                 };
-                Poll::Ready(Some(Ok((head, body))))
+                Poll::Ready(Some(Ok((head, Some(body)))))
             } else {
                 unreachable!("poll_msg shouldn't be called if no inflight");
             };
 
             // Since in_flight finished, remove it
             this.in_flight.set(None);
+            this.informational_rx = None;
             ret
         }
 
@@ -561,7 +597,10 @@ cfg_server! {
             *req.headers_mut() = msg.headers;
             *req.version_mut() = msg.version;
             *req.extensions_mut() = msg.extensions;
+            let (informational_tx, informational_rx) = mpsc::channel(1);
+            assert!(req.extensions_mut().insert(InformationalSender(informational_tx)).is_none());
             let fut = self.service.call(req);
+            self.informational_rx = Some(informational_rx);
             self.in_flight.set(Some(fut));
             Ok(())
         }
@@ -607,7 +646,7 @@ cfg_client! {
         fn poll_msg(
             mut self: Pin<&mut Self>,
             cx: &mut Context<'_>,
-        ) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), Infallible>>> {
+        ) -> Poll<Option<Result<(Self::PollItem, Option<Self::PollBody>), Infallible>>> {
             let mut this = self.as_mut();
             debug_assert!(!this.rx_closed);
             match this.rx.poll_recv(cx) {
@@ -627,7 +666,7 @@ cfg_client! {
                                 extensions: parts.extensions,
                             };
                             this.callback = Some(cb);
-                            Poll::Ready(Some(Ok((head, body))))
+                            Poll::Ready(Some(Ok((head, Some(body)))))
                         }
                     }
                 }

src/proto/h1/mod.rs

@@ -33,7 +33,8 @@ pub(crate) trait Http1Transaction {
     #[cfg(feature = "tracing")]
     const LOG: &'static str;
     fn parse(bytes: &mut BytesMut, ctx: ParseContext<'_>) -> ParseResult<Self::Incoming>;
-    fn encode(enc: Encode<'_, Self::Outgoing>, dst: &mut Vec<u8>) -> crate::Result<Encoder>;
+    fn encode(enc: Encode<'_, Self::Outgoing>, dst: &mut Vec<u8>)
+        -> crate::Result<Option<Encoder>>;
 
     fn on_error(err: &crate::Error) -> Option<MessageHead<Self::Outgoing>>;
 

src/proto/h1/role.rs

@@ -111,7 +111,7 @@ fn is_complete_fast(bytes: &[u8], prev_len: usize) -> bool {
 pub(super) fn encode_headers<T>(
     enc: Encode<'_, T::Outgoing>,
     dst: &mut Vec<u8>,
-) -> crate::Result<Encoder>
+) -> crate::Result<Option<Encoder>>
 where
     T: Http1Transaction,
 {
@@ -358,7 +358,10 @@ impl Http1Transaction for Server {
         }))
     }
 
-    fn encode(mut msg: Encode<'_, Self::Outgoing>, dst: &mut Vec<u8>) -> crate::Result<Encoder> {
+    fn encode(
+        msg: Encode<'_, Self::Outgoing>,
+        dst: &mut Vec<u8>,
+    ) -> crate::Result<Option<Encoder>> {
         trace!(
             "Server::encode status={:?}, body={:?}, req_method={:?}",
             msg.head.subject,
@@ -368,25 +371,19 @@ impl Http1Transaction for Server {
 
         let mut wrote_len = false;
 
-        // hyper currently doesn't support returning 1xx status codes as a Response
-        // This is because Service only allows returning a single Response, and
-        // so if you try to reply with a e.g. 100 Continue, you have no way of
-        // replying with the latter status code response.
-        let (ret, is_last) = if msg.head.subject == StatusCode::SWITCHING_PROTOCOLS {
-            (Ok(()), true)
+        let informational = msg.head.subject.is_informational();
+
+        let is_last = if msg.head.subject == StatusCode::SWITCHING_PROTOCOLS {
+            true
         } else if msg.req_method == &Some(Method::CONNECT) && msg.head.subject.is_success() {
             // Sending content-length or transfer-encoding header on 2xx response
             // to CONNECT is forbidden in RFC 7231.
             wrote_len = true;
-            (Ok(()), true)
-        } else if msg.head.subject.is_informational() {
-            warn!("response with 1xx status code not supported");
-            *msg.head = MessageHead::default();
-            msg.head.subject = StatusCode::INTERNAL_SERVER_ERROR;
-            msg.body = None;
-            (Err(crate::Error::new_user_unsupported_status_code()), true)
+            true
+        } else if informational {
+            false
         } else {
-            (Ok(()), !msg.keep_alive)
+            !msg.keep_alive
         };
 
         // In some error cases, we don't know about the invalid message until already
@@ -444,6 +441,7 @@ impl Http1Transaction for Server {
             }
             orig_headers => orig_headers,
         };
+
         let encoder = if let Some(orig_headers) = orig_headers {
             Self::encode_headers_with_original_case(
                 msg,
@@ -457,7 +455,11 @@ impl Http1Transaction for Server {
             Self::encode_headers_with_lower_case(msg, dst, is_last, orig_len, wrote_len)?
         };
 
-        ret.map(|()| encoder)
+        // If we're senting a 1xx informational response, it won't have a body,
+        // so we'll return `None` here.  Additionally, that will tell
+        // `Conn::write_head` to stay in the `Writing::Init` state since we
+        // haven't yet sent the final response.
+        Ok(if informational { None } else { Some(encoder) })
     }
 
     fn on_error(err: &crate::Error) -> Option<MessageHead<Self::Outgoing>> {
@@ -1168,7 +1170,10 @@ impl Http1Transaction for Client {
         }
     }
 
-    fn encode(msg: Encode<'_, Self::Outgoing>, dst: &mut Vec<u8>) -> crate::Result<Encoder> {
+    fn encode(
+        msg: Encode<'_, Self::Outgoing>,
+        dst: &mut Vec<u8>,
+    ) -> crate::Result<Option<Encoder>> {
         trace!(
             "Client::encode method={:?}, body={:?}",
             msg.head.subject.0,
@@ -1214,7 +1219,7 @@ impl Http1Transaction for Client {
         extend(dst, b"\r\n");
         msg.head.headers.clear(); //TODO: remove when switching to drain()
 
-        Ok(body)
+        Ok(Some(body))
     }
 
     fn on_error(_err: &crate::Error) -> Option<MessageHead<Self::Outgoing>> {