Merge pull request #176 from palantir/async-service

Make Service async fn-compatible

.circleci/config.yml

@@ -22,7 +22,7 @@ version: 2
 jobs:
   build:
     docker:
-      - image: rust:1.65.0
+      - image: rust:1.75.0
     environment:
       RUSTFLAGS: -D warnings
     steps:

changelog/@unreleased/pr-176.v2.yml

@@ -0,0 +1,5 @@
+type: break
+break:
+  description: The `Service` trait now supports `async fn call`.
+  links:
+  - https://github.com/palantir/conjure-rust-runtime/pull/176

conjure-runtime/src/blocking/client.rs

@@ -76,7 +76,6 @@ impl<T, B> conjure_http::client::Client for Client<T>
 where
     T: Service<Request<RawBody>, Response = Response<B>> + 'static + Sync + Send,
     T::Error: Into<Box<dyn error::Error + Sync + Send>>,
-    T::Future: Send,
     B: http_body::Body<Data = Bytes> + 'static + Send,
     B::Error: Into<Box<dyn error::Error + Sync + Send>>,
 {

conjure-runtime/src/client.rs

@@ -147,7 +147,6 @@ impl<T, B> AsyncClient for Client<T>
 where
     T: Service<http::Request<RawBody>, Response = http::Response<B>> + 'static + Sync + Send,
     T::Error: Into<Box<dyn error::Error + Sync + Send>>,
-    T::Future: Send,
     B: http_body::Body<Data = Bytes> + 'static + Send,
     B::Error: Into<Box<dyn error::Error + Sync + Send>>,
 {
@@ -159,8 +158,6 @@ where
         &self,
         request: Request<AsyncRequestBody<'_, Self::BodyWriter>>,
     ) -> Result<Response<Self::ResponseBody>, Error> {
-        // split into 2 statements to avoid holding onto the state while awaiting the future
-        let future = self.state.load().service.call(request);
-        future.await
+        self.state.load().service.call(request).await
     }
 }

conjure-runtime/src/raw/default.rs

@@ -18,10 +18,9 @@ use crate::service::tls_metrics::{TlsMetricsLayer, TlsMetricsService};
 use crate::Builder;
 use bytes::Bytes;
 use conjure_error::Error;
-use futures::ready;
 use http::{HeaderMap, Request, Response};
 use http_body::{Body, SizeHint};
-use hyper::client::{HttpConnector, ResponseFuture};
+use hyper::client::HttpConnector;
 use hyper::Client;
 use hyper_rustls::{HttpsConnector, HttpsConnectorBuilder};
 use pin_project::pin_project;
@@ -30,7 +29,6 @@ use rustls_pemfile::Item;
 use std::error;
 use std::fmt;
 use std::fs::File;
-use std::future::Future;
 use std::io::BufReader;
 use std::marker::PhantomPinned;
 use std::path::Path;
@@ -162,13 +160,18 @@ pub struct DefaultRawClient(Client<ConjureConnector, RawBody>);
 impl Service<Request<RawBody>> for DefaultRawClient {
     type Response = Response<DefaultRawBody>;
     type Error = DefaultRawError;
-    type Future = DefaultRawFuture;
 
-    fn call(&self, req: Request<RawBody>) -> Self::Future {
-        DefaultRawFuture {
-            future: self.0.request(req),
-            _p: PhantomPinned,
-        }
+    async fn call(&self, req: Request<RawBody>) -> Result<Self::Response, Self::Error> {
+        self.0
+            .request(req)
+            .await
+            .map(|r| {
+                r.map(|inner| DefaultRawBody {
+                    inner,
+                    _p: PhantomPinned,
+                })
+            })
+            .map_err(DefaultRawError)
     }
 }
 
@@ -214,29 +217,6 @@ impl Body for DefaultRawBody {
     }
 }
 
-/// The future type used by `DefaultRawClient`.
-#[pin_project]
-pub struct DefaultRawFuture {
-    #[pin]
-    future: ResponseFuture,
-    #[pin]
-    _p: PhantomPinned,
-}
-
-impl Future for DefaultRawFuture {
-    type Output = Result<Response<DefaultRawBody>, DefaultRawError>;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let response = ready!(self.project().future.poll(cx)).map_err(DefaultRawError)?;
-        let response = response.map(|inner| DefaultRawBody {
-            inner,
-            _p: PhantomPinned,
-        });
-
-        Poll::Ready(Ok(response))
-    }
-}
-
 /// The error type used by `DefaultRawClient`.
 #[derive(Debug)]
 pub struct DefaultRawError(hyper::Error);

conjure-runtime/src/raw/mod.rs

@@ -53,11 +53,9 @@ pub trait Service<R> {
     type Response;
     /// The error type returned by the service.
     type Error;
-    /// The future type returned by the service.
-    type Future: Future<Output = Result<Self::Response, Self::Error>>;
 
     /// Asynchronously perform the request.
-    fn call(&self, req: R) -> Self::Future;
+    fn call(&self, req: R) -> impl Future<Output = Result<Self::Response, Self::Error>> + Send;
 }
 
 impl<R, T> Service<R> for Arc<T>
@@ -66,9 +64,8 @@ where
 {
     type Response = T::Response;
     type Error = T::Error;
-    type Future = T::Future;
 
-    fn call(&self, req: R) -> Self::Future {
+    fn call(&self, req: R) -> impl Future<Output = Result<T::Response, T::Error>> {
         (**self).call(req)
     }
 }

conjure-runtime/src/service/gzip.rs

@@ -23,7 +23,6 @@ use http_body::{Body, SizeHint};
 use once_cell::sync::Lazy;
 use pin_project::pin_project;
 use std::error::Error;
-use std::future::Future;
 use std::io::Write;
 use std::pin::Pin;
 use std::task::{Context, Poll};
@@ -50,41 +49,20 @@ pub struct GzipService<S> {
 
 impl<S, B1, B2> Service<Request<B1>> for GzipService<S>
 where
-    S: Service<Request<B1>, Response = Response<B2>>,
+    S: Service<Request<B1>, Response = Response<B2>> + Sync + Send,
+    B1: Sync + Send,
     B2: Body<Data = Bytes>,
     B2::Error: Into<Box<dyn Error + Sync + Send>>,
 {
     type Response = Response<DecodedBody<B2>>;
     type Error = S::Error;
-    type Future = GzipFuture<S::Future>;
 
-    fn call(&self, mut req: Request<B1>) -> Self::Future {
+    async fn call(&self, mut req: Request<B1>) -> Result<Self::Response, Self::Error> {
         if let Entry::Vacant(e) = req.headers_mut().entry(ACCEPT_ENCODING) {
             e.insert(GZIP.clone());
         }
 
-        GzipFuture {
-            future: self.inner.call(req),
-        }
-    }
-}
-
-#[pin_project]
-pub struct GzipFuture<F> {
-    #[pin]
-    future: F,
-}
-
-impl<F, E, B> Future for GzipFuture<F>
-where
-    F: Future<Output = Result<Response<B>, E>>,
-    B: Body<Data = Bytes>,
-    B::Error: Into<Box<dyn Error + Sync + Send>>,
-{
-    type Output = Result<Response<DecodedBody<B>>, E>;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let response = ready!(self.project().future.poll(cx))?;
+        let response = self.inner.call(req).await?;
         let (mut parts, body) = response.into_parts();
 
         let decoder = match parts.headers.get(CONTENT_ENCODING) {
@@ -102,7 +80,7 @@ where
             done: false,
         };
 
-        Poll::Ready(Ok(Response::from_parts(parts, body)))
+        Ok(Response::from_parts(parts, body))
     }
 }