refactor(proxy/http): http/1 client is a Service<T>

this is a small mechanical refactor to the http/1 client.

our http/2 and "orig_proto" clients are tower services. our http/1
client, on the other hand, exposes a concrete inherent method `request`.

to be consistent, this changes our http client to treat this http/1
client as a service as well.

Signed-off-by: katelyn martin <kate@buoyant.io>

Author: cratelyn

linkerd/proxy/http/src/client.rs

@@ -136,7 +136,7 @@ where
         match self {
             Self::H2(ref mut svc) => svc.poll_ready(cx).map_err(Into::into),
             Self::OrigProtoUpgrade(ref mut svc) => svc.poll_ready(cx),
-            Self::Http1(_) => Poll::Ready(Ok(())),
+            Self::Http1(ref mut svc) => svc.poll_ready(cx),
         }
     }
 
@@ -155,7 +155,7 @@ where
             debug!(headers = ?req.headers());
 
             match self {
-                Self::Http1(ref mut h1) => h1.request(req),
+                Self::Http1(ref mut svc) => svc.call(req),
                 Self::OrigProtoUpgrade(ref mut svc) => svc.call(req),
                 Self::H2(ref mut svc) => Box::pin(
                     svc.call(req)

linkerd/proxy/http/src/h1.rs

@@ -61,7 +61,7 @@ impl<C: Clone, T: Clone, B> Clone for Client<C, T, B> {
 
 type RspFuture = Pin<Box<dyn Future<Output = Result<http::Response<BoxBody>>> + Send + 'static>>;
 
-impl<C, T, B> Client<C, T, B>
+impl<C, T, B> tower::Service<http::Request<B>> for Client<C, T, B>
 where
     T: Clone + Send + Sync + 'static,
     C: MakeConnection<(crate::Variant, T)> + Clone + Send + Sync + 'static,
@@ -71,7 +71,15 @@ where
     B::Data: Send,
     B::Error: Into<Error> + Send + Sync,
 {
-    pub(crate) fn request(&mut self, mut req: http::Request<B>) -> RspFuture {
+    type Response = http::Response<BoxBody>;
+    type Error = Error;
+    type Future = RspFuture;
+
+    fn poll_ready(&mut self, _: &mut std::task::Context<'_>) -> std::task::Poll<Result<()>> {
+        std::task::Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, mut req: http::Request<B>) -> RspFuture {
         // Marked by `upgrade`.
         let upgrade = req.extensions_mut().remove::<Http11Upgrade>();
         let is_http_connect = req.method() == http::Method::CONNECT;

linkerd/proxy/http/src/orig_proto.rs

@@ -67,7 +67,14 @@ where
 
     #[inline]
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.h2.poll_ready(cx).map_err(downgrade_h2_error)
+        let Self { http1, h2 } = self;
+
+        match http1.poll_ready(cx) {
+            Poll::Ready(Ok(())) => {}
+            poll => return poll,
+        }
+
+        h2.poll_ready(cx).map_err(downgrade_h2_error)
     }
 
     fn call(&mut self, mut req: http::Request<B>) -> Self::Future {
@@ -78,7 +85,7 @@ where
             .is_some()
         {
             debug!("Skipping orig-proto upgrade due to HTTP/1.1 upgrade");
-            return Box::pin(self.http1.request(req).map_ok(|rsp| rsp.map(BoxBody::new)));
+            return Box::pin(self.http1.call(req).map_ok(|rsp| rsp.map(BoxBody::new)));
         }
 
         let orig_version = req.version();