hacky support for 1xx informational responses

It's ugly, but here's what I did: - Add `service::InformationalSender`, a type which wraps a `futures_channel::mspc::Sender<Response<()>>`. This may be added as an extension to a `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 incoming 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 informatinal responses in the HTTP/2 server. TODO hyperium#2: Come up with a less hacky API. Signed-off-by: Joel Dice <[email protected]>
dicej · Dec 17, 2024 · 0fe9a87 · 0fe9a87
1 parent a3bda62
commit 0fe9a87
Show file tree

Hide file tree

Showing 6 changed files with 87 additions and 37 deletions.
diff --git a/src/proto/h1/conn.rs b/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);

diff --git a/src/proto/h1/dispatch.rs b/src/proto/h1/dispatch.rs
@@ -8,15 +8,19 @@ 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;
 use crate::proto::{BodyLength, Conn, Dispatched, MessageHead, RequestHead};
+#[cfg(feature = "server")]
+use crate::service::InformationalSender;
 use crate::upgrade::OnUpgrade;
 
 pub(crate) struct Dispatcher<D, Bs: Body, I, T> {
@@ -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)))))
                         }
                     }
                 }

diff --git a/src/proto/h1/mod.rs b/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>>;
 

diff --git a/src/proto/h1/role.rs b/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,7 @@ impl Http1Transaction for Server {
             Self::encode_headers_with_lower_case(msg, dst, is_last, orig_len, wrote_len)?
         };
 
-        ret.map(|()| encoder)
+        Ok(if informational { None } else { Some(encoder) })
     }
 
     fn on_error(err: &crate::Error) -> Option<MessageHead<Self::Outgoing>> {
@@ -1168,7 +1166,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 +1215,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>> {

diff --git a/src/service/http.rs b/src/service/http.rs
@@ -47,6 +47,12 @@ where
 {
 }
 
+/// Request extension type for sending one or more informational responses prior
+/// to the final response.
+#[cfg(all(feature = "server", any(feature = "http1", feature = "http2")))]
+#[derive(Clone, Debug)]
+pub struct InformationalSender(pub futures_channel::mpsc::Sender<Response<()>>);
+
 mod sealed {
     pub trait Sealed<T> {}
 }
diff --git a/src/service/mod.rs b/src/service/mod.rs
@@ -28,3 +28,6 @@ mod util;
 pub use self::http::HttpService;
 pub use self::service::Service;
 pub use self::util::service_fn;
+
+#[cfg(all(feature = "server", any(feature = "http1", feature = "http2")))]
+pub use self::http::InformationalSender;