Move the BoxCloneSyncService to mas-http

crates/http/src/client.rs

@@ -27,8 +27,8 @@ use hyper::{
 use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder};
 use thiserror::Error;
 use tower::{
-    util::{BoxCloneService, MapErrLayer, MapResponseLayer},
-    Layer, Service, ServiceExt,
+    util::{MapErrLayer, MapResponseLayer},
+    Layer, Service,
 };
 
 use crate::{
@@ -36,7 +36,7 @@ use crate::{
         client::{ClientLayer, ClientResponse},
         otel::{TraceDns, TraceLayer},
     },
-    BoxError,
+    BoxCloneSyncService, BoxError,
 };
 
 #[cfg(all(not(feature = "webpki-roots"), not(feature = "native-roots")))]
@@ -237,7 +237,7 @@ where
 pub async fn client<B, E>(
     operation: &'static str,
 ) -> Result<
-    BoxCloneService<Request<B>, Response<BoxBody<bytes::Bytes, ClientError>>, ClientError>,
+    BoxCloneSyncService<Request<B>, Response<BoxBody<bytes::Bytes, ClientError>>, ClientError>,
     ClientInitError,
 >
 where
@@ -254,7 +254,7 @@ where
         }),
         ClientLayer::new(operation),
     );
-    let client = layer.layer(client).boxed_clone();
+    let client = BoxCloneSyncService::new(layer.layer(client));
 
     Ok(client)
 }

crates/http/src/layers/client.rs

@@ -16,17 +16,18 @@ use std::{marker::PhantomData, time::Duration};
 
 use http::{header::USER_AGENT, HeaderValue, Request, Response};
 use tower::{
-    limit::ConcurrencyLimitLayer, timeout::TimeoutLayer, util::BoxCloneService, Layer, Service,
-    ServiceExt,
+    limit::{ConcurrencyLimit, ConcurrencyLimitLayer},
+    timeout::{Timeout, TimeoutLayer},
+    Layer, Service,
 };
 use tower_http::{
-    decompression::{DecompressionBody, DecompressionLayer},
-    follow_redirect::FollowRedirectLayer,
-    set_header::SetRequestHeaderLayer,
+    decompression::{Decompression, DecompressionBody, DecompressionLayer},
+    follow_redirect::{FollowRedirect, FollowRedirectLayer},
+    set_header::{SetRequestHeader, SetRequestHeaderLayer},
 };
 
 use super::otel::TraceLayer;
-use crate::BoxError;
+use crate::{otel::TraceHttpClient, BoxError};
 
 static MAS_USER_AGENT: HeaderValue =
     HeaderValue::from_static("matrix-authentication-service/0.0.1");
@@ -47,17 +48,27 @@ impl<B> ClientLayer<B> {
     }
 }
 
+#[allow(dead_code)]
 pub type ClientResponse<B> = Response<DecompressionBody<B>>;
 
 impl<ReqBody, ResBody, S, E> Layer<S> for ClientLayer<ReqBody>
 where
-    S: Service<Request<ReqBody>, Response = Response<ResBody>, Error = E> + Clone + Send + 'static,
+    S: Service<Request<ReqBody>, Response = Response<ResBody>, Error = E>
+        + Clone
+        + Send
+        + Sync
+        + 'static,
     ReqBody: http_body::Body + Default + Send + 'static,
     ResBody: http_body::Body + Sync + Send + 'static,
     S::Future: Send + 'static,
     E: Into<BoxError>,
 {
-    type Service = BoxCloneService<Request<ReqBody>, ClientResponse<ResBody>, BoxError>;
+    type Service = Decompression<
+        SetRequestHeader<
+            TraceHttpClient<ConcurrencyLimit<FollowRedirect<TraceHttpClient<Timeout<S>>>>>,
+            HeaderValue,
+        >,
+    >;
 
     fn layer(&self, inner: S) -> Self::Service {
         // Note that most layers here just forward the error type. Two notables
@@ -78,6 +89,5 @@ where
             TimeoutLayer::new(Duration::from_secs(10)),
         )
             .layer(inner)
-            .boxed_clone()
     }
 }

crates/http/src/lib.rs

@@ -28,6 +28,7 @@
 mod client;
 mod ext;
 mod layers;
+mod service;
 
 #[cfg(feature = "client")]
 pub use self::client::{client, make_traced_connector, make_untraced_client, ClientInitError};
@@ -44,6 +45,7 @@ pub use self::{
         otel,
         server::ServerLayer,
     },
+    service::{BoxCloneSyncService, HttpService},
 };
 
 pub(crate) type BoxError = Box<dyn std::error::Error + Send + Sync>;

crates/http/src/service.rs

@@ -0,0 +1,104 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::{
+    fmt,
+    task::{Context, Poll},
+};
+
+use bytes::Bytes;
+use futures_util::future::BoxFuture;
+use tower::{BoxError, Service, ServiceExt};
+
+/// Type for the underlying HTTP service.
+///
+/// Allows implementors to use different libraries that provide a [`Service`]
+/// that implements [`Clone`] + [`Send`] + [`Sync`].
+pub type HttpService = BoxCloneSyncService<http::Request<Bytes>, http::Response<Bytes>, BoxError>;
+
+impl fmt::Debug for HttpService {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt.debug_struct("HttpService").finish()
+    }
+}
+
+/// A [`Clone`] + [`Send`] + [`Sync`] boxed [`Service`].
+///
+/// [`BoxCloneSyncService`] turns a service into a trait object, allowing the
+/// response future type to be dynamic, and allowing the service to be cloned.
+#[allow(clippy::type_complexity)]
+pub struct BoxCloneSyncService<T, U, E>(
+    Box<
+        dyn CloneSyncService<T, Response = U, Error = E, Future = BoxFuture<'static, Result<U, E>>>,
+    >,
+);
+
+impl<T, U, E> BoxCloneSyncService<T, U, E> {
+    /// Create a new `BoxCloneSyncService`.
+    pub fn new<S>(inner: S) -> Self
+    where
+        S: Service<T, Response = U, Error = E> + Clone + Send + Sync + 'static,
+        S::Future: Send + 'static,
+    {
+        let inner = inner.map_future(|f| Box::pin(f) as _);
+        Self(Box::new(inner))
+    }
+}
+
+impl<T, U, E> Service<T> for BoxCloneSyncService<T, U, E> {
+    type Response = U;
+    type Error = E;
+    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+    #[inline]
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.0.poll_ready(cx)
+    }
+
+    #[inline]
+    fn call(&mut self, request: T) -> Self::Future {
+        self.0.call(request)
+    }
+}
+
+impl<T, U, E> Clone for BoxCloneSyncService<T, U, E> {
+    fn clone(&self) -> Self {
+        Self(self.0.clone_sync_box())
+    }
+}
+
+trait CloneSyncService<R>: Service<R> + Send + Sync {
+    fn clone_sync_box(
+        &self,
+    ) -> Box<
+        dyn CloneSyncService<
+            R,
+            Response = Self::Response,
+            Error = Self::Error,
+            Future = Self::Future,
+        >,
+    >;
+}
+
+impl<R, T> CloneSyncService<R> for T
+where
+    T: Service<R> + Send + Sync + Clone + 'static,
+{
+    fn clone_sync_box(
+        &self,
+    ) -> Box<dyn CloneSyncService<R, Response = T::Response, Error = T::Error, Future = T::Future>>
+    {
+        Box::new(self.clone())
+    }
+}