feat(proto,server,service): completely remove server module's reliance on tower::Service

tomkarw · tomkarw · commit e7c6b5a54939 · 2022-08-18T20:49:16.000+02:00
diff --git a/src/proto/h1/dispatch.rs b/src/proto/h1/dispatch.rs
@@ -31,14 +31,13 @@ pub(crate) trait Dispatch {
         cx: &mut task::Context<'_>,
     ) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), Self::PollError>>>;
     fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()>;
-    fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), ()>>;
     fn should_poll(&self) -> bool;
 }
 
 cfg_server! {
-    use crate::service::TowerHttpService;
+    use crate::service::HttpService;
 
-    pub(crate) struct Server<S: TowerHttpService<B>, B> {
+    pub(crate) struct Server<S: HttpService<B>, B> {
         in_flight: Pin<Box<Option<S::Future>>>,
         pub(crate) service: S,
     }
@@ -235,15 +234,6 @@ where
     }
 
     fn poll_read_head(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
-        // can dispatch receive, or does it still care about, an incoming message?
-        match ready!(self.dispatch.poll_ready(cx)) {
-            Ok(()) => (),
-            Err(()) => {
-                trace!("dispatch no longer receiving messages");
-                self.close();
-                return Poll::Ready(Ok(()));
-            }
-        }
         // dispatch is ready for a message, try to read one
         match ready!(self.conn.poll_read_head(cx)) {
             Some(Ok((mut head, body_len, wants))) => {
@@ -454,7 +444,7 @@ impl<'a, T> Drop for OptGuard<'a, T> {
 cfg_server! {
     impl<S, B> Server<S, B>
     where
-        S: TowerHttpService<B>,
+        S: HttpService<B>,
     {
         pub(crate) fn new(service: S) -> Server<S, B> {
             Server {
@@ -469,11 +459,11 @@ cfg_server! {
     }
 
     // Service is never pinned
-    impl<S: TowerHttpService<B>, B> Unpin for Server<S, B> {}
+    impl<S: HttpService<B>, B> Unpin for Server<S, B> {}
 
     impl<S, Bs> Dispatch for Server<S, Body>
     where
-        S: TowerHttpService<Body, ResBody = Bs>,
+        S: HttpService<Body, ResBody = Bs>,
         S::Error: Into<Box<dyn StdError + Send + Sync>>,
         Bs: HttpBody,
     {
@@ -519,17 +509,6 @@ cfg_server! {
             Ok(())
         }
 
-        fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), ()>> {
-            if self.in_flight.is_some() {
-                Poll::Pending
-            } else {
-                self.service.poll_ready(cx).map_err(|_e| {
-                    // FIXME: return error value.
-                    trace!("service closed");
-                })
-            }
-        }
-
         fn should_poll(&self) -> bool {
             self.in_flight.is_some()
         }
@@ -631,19 +610,6 @@ cfg_client! {
             }
         }
 
-        fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), ()>> {
-            match self.callback {
-                Some(ref mut cb) => match cb.poll_canceled(cx) {
-                    Poll::Ready(()) => {
-                        trace!("callback receiver has dropped");
-                        Poll::Ready(Err(()))
-                    }
-                    Poll::Pending => Poll::Ready(Ok(())),
-                },
-                None => Poll::Ready(Err(())),
-            }
-        }
-
         fn should_poll(&self) -> bool {
             self.callback.is_none()
         }
diff --git a/src/proto/h2/server.rs b/src/proto/h2/server.rs
@@ -20,7 +20,7 @@ use crate::headers;
 use crate::proto::h2::ping::Recorder;
 use crate::proto::h2::{H2Upgraded, UpgradedSendStream};
 use crate::proto::Dispatched;
-use crate::service::TowerHttpService;
+use crate::service::HttpService;
 
 use crate::upgrade::{OnUpgrade, Pending, Upgraded};
 use crate::{Body, Response};
@@ -76,7 +76,7 @@ impl Default for Config {
 pin_project! {
     pub(crate) struct Server<T, S, B, E>
     where
-        S: TowerHttpService<Body>,
+        S: HttpService<Body>,
         B: HttpBody,
     {
         exec: E,
@@ -109,7 +109,7 @@ where
 impl<T, S, B, E> Server<T, S, B, E>
 where
     T: AsyncRead + AsyncWrite + Unpin,
-    S: TowerHttpService<Body, ResBody = B>,
+    S: HttpService<Body, ResBody = B>,
     S::Error: Into<Box<dyn StdError + Send + Sync>>,
     B: HttpBody + 'static,
     E: ConnStreamExec<S::Future, B>,
@@ -181,7 +181,7 @@ where
 impl<T, S, B, E> Future for Server<T, S, B, E>
 where
     T: AsyncRead + AsyncWrite + Unpin,
-    S: TowerHttpService<Body, ResBody = B>,
+    S: HttpService<Body, ResBody = B>,
     S::Error: Into<Box<dyn StdError + Send + Sync>>,
     B: HttpBody + 'static,
     E: ConnStreamExec<S::Future, B>,
@@ -236,46 +236,14 @@ where
         exec: &mut E,
     ) -> Poll<crate::Result<()>>
     where
-        S: TowerHttpService<Body, ResBody = B>,
+        S: HttpService<Body, ResBody = B>,
         S::Error: Into<Box<dyn StdError + Send + Sync>>,
         E: ConnStreamExec<S::Future, B>,
     {
         if self.closing.is_none() {
             loop {
                 self.poll_ping(cx);
 
-                // Check that the service is ready to accept a new request.
-                //
-                // - If not, just drive the connection some.
-                // - If ready, try to accept a new request from the connection.
-                match service.poll_ready(cx) {
-                    Poll::Ready(Ok(())) => (),
-                    Poll::Pending => {
-                        // use `poll_closed` instead of `poll_accept`,
-                        // in order to avoid accepting a request.
-                        ready!(self.conn.poll_closed(cx).map_err(crate::Error::new_h2))?;
-                        trace!("incoming connection complete");
-                        return Poll::Ready(Ok(()));
-                    }
-                    Poll::Ready(Err(err)) => {
-                        let err = crate::Error::new_user_service(err);
-                        debug!("service closed: {}", err);
-
-                        let reason = err.h2_reason();
-                        if reason == Reason::NO_ERROR {
-                            // NO_ERROR is only used for graceful shutdowns...
-                            trace!("interpreting NO_ERROR user error as graceful_shutdown");
-                            self.conn.graceful_shutdown();
-                        } else {
-                            trace!("abruptly shutting down with {:?}", reason);
-                            self.conn.abrupt_shutdown(reason);
-                        }
-                        self.closing = Some(err);
-                        break;
-                    }
-                }
-
-                // When the service is ready, accepts an incoming request.
                 match ready!(self.conn.poll_accept(cx)) {
                     Some(Ok((req, mut respond))) => {
                         trace!("incoming request");
diff --git a/src/server/conn.rs b/src/server/conn.rs
@@ -72,7 +72,7 @@ cfg_feature! {
     use crate::common::Never;
     use crate::common::exec::{ConnStreamExec, Exec};
     use crate::proto;
-    use crate::service::TowerHttpService;
+    use crate::service::HttpService;
 
     pub(super) use self::upgrades::UpgradeableConnection;
 }
@@ -123,7 +123,7 @@ pin_project! {
     #[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
     pub struct Connection<T, S, E = Exec>
     where
-        S: TowerHttpService<Body>,
+        S: HttpService<Body>,
     {
         pub(super) conn: Option<ProtoServer<T, S::ResBody, S, E>>,
         fallback: Fallback<E>,
@@ -151,7 +151,7 @@ pin_project! {
     #[project = ProtoServerProj]
     pub(super) enum ProtoServer<T, B, S, E = Exec>
     where
-        S: TowerHttpService<Body>,
+        S: HttpService<Body>,
         B: HttpBody,
     {
         H1 {
@@ -598,7 +598,7 @@ impl<E> Http<E> {
     /// ```
     pub fn serve_connection<S, I, Bd>(&self, io: I, service: S) -> Connection<I, S, E>
     where
-        S: TowerHttpService<Body, ResBody = Bd>,
+        S: HttpService<Body, ResBody = Bd>,
         S::Error: Into<Box<dyn StdError + Send + Sync>>,
         Bd: HttpBody + 'static,
         Bd::Error: Into<Box<dyn StdError + Send + Sync>>,
@@ -678,7 +678,7 @@ impl<E> Http<E> {
 #[cfg(any(feature = "http1", feature = "http2"))]
 impl<I, B, S, E> Connection<I, S, E>
 where
-    S: TowerHttpService<Body, ResBody = B>,
+    S: HttpService<Body, ResBody = B>,
     S::Error: Into<Box<dyn StdError + Send + Sync>>,
     I: AsyncRead + AsyncWrite + Unpin,
     B: HttpBody + 'static,
@@ -848,7 +848,7 @@ where
 #[cfg(any(feature = "http1", feature = "http2"))]
 impl<I, B, S, E> Future for Connection<I, S, E>
 where
-    S: TowerHttpService<Body, ResBody = B>,
+    S: HttpService<Body, ResBody = B>,
     S::Error: Into<Box<dyn StdError + Send + Sync>>,
     I: AsyncRead + AsyncWrite + Unpin + 'static,
     B: HttpBody + 'static,
@@ -895,7 +895,7 @@ where
 #[cfg(any(feature = "http1", feature = "http2"))]
 impl<I, S> fmt::Debug for Connection<I, S>
 where
-    S: TowerHttpService<Body>,
+    S: HttpService<Body>,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Connection").finish()
@@ -928,7 +928,7 @@ impl Default for ConnectionMode {
 impl<T, B, S, E> Future for ProtoServer<T, B, S, E>
 where
     T: AsyncRead + AsyncWrite + Unpin,
-    S: TowerHttpService<Body, ResBody = B>,
+    S: HttpService<Body, ResBody = B>,
     S::Error: Into<Box<dyn StdError + Send + Sync>>,
     B: HttpBody + 'static,
     B::Error: Into<Box<dyn StdError + Send + Sync>>,
@@ -963,14 +963,14 @@ mod upgrades {
     #[allow(missing_debug_implementations)]
     pub struct UpgradeableConnection<T, S, E>
     where
-        S: TowerHttpService<Body>,
+        S: HttpService<Body>,
     {
         pub(super) inner: Connection<T, S, E>,
     }
 
     impl<I, B, S, E> UpgradeableConnection<I, S, E>
     where
-        S: TowerHttpService<Body, ResBody = B>,
+        S: HttpService<Body, ResBody = B>,
         S::Error: Into<Box<dyn StdError + Send + Sync>>,
         I: AsyncRead + AsyncWrite + Unpin,
         B: HttpBody + 'static,
@@ -988,7 +988,7 @@ mod upgrades {
 
     impl<I, B, S, E> Future for UpgradeableConnection<I, S, E>
     where
-        S: TowerHttpService<Body, ResBody = B>,
+        S: HttpService<Body, ResBody = B>,
         S::Error: Into<Box<dyn StdError + Send + Sync>>,
         I: AsyncRead + AsyncWrite + Unpin + Send + 'static,
         B: HttpBody + 'static,
diff --git a/src/service/mod.rs b/src/service/mod.rs
@@ -23,9 +23,9 @@
 
 mod http;
 mod service;
-mod tower_http;
 mod util;
 
-pub(super) use self::tower_http::TowerHttpService;
+#[cfg(all(any(feature = "http1", feature = "http2"), feature = "server"))]
+pub(super) use self::http::HttpService;
 
 pub use self::util::service_fn;
diff --git a/src/service/tower_http.rs b/src/service/tower_http.rs
