Repository: tower-rs/tower
Pull request #436 (2020-04-14 17:17:27)

Refactor github actions


Patch:
diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml
index b7816b5..4194fc2 100644
--- a/.github/workflows/CI.yml
+++ b/.github/workflows/CI.yml
@@ -7,161 +7,125 @@ on:
   pull_request: {}
 
 jobs:
-  test-workspace:
-
-    runs-on: ${{ matrix.os }}
-
+  check:
+    # Run `cargo check` first to ensure that the pushed code at least compiles.
+    runs-on: ubuntu-latest
     strategy:
       matrix:
-        os: [ubuntu-18.04, windows-2019, macOS-10.14]
-        rust: [stable, nightly]
-
+        rust: [stable, 1.39.0]
     steps:
-      - name: Checkout code
-        uses: actions/checkout@v1
-      - name: Install rust toolchain
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: ${{ matrix.rust }}
-          override: true
-      - name: Build
-        uses: actions-rs/cargo@v1
-        with:
-          command: check
-          args: --verbose
-      - name: Run tests
-        uses: actions-rs/cargo@v1
-        env:
-          CI: 'True'
-          RUSTFLAGS: '-D warnings'
-        with:
-          command: test
-          args: --verbose
-
-  test:
-
+    - uses: actions/checkout@master
+    - uses: actions-rs/toolchain@v1
+      with:
+        toolchain: ${{ matrix.rust }}
+        profile: minimal
+    - name: Check
+      uses: actions-rs/cargo@v1
+      with:
+        command: check
+        args: --all --all-targets --all-features
+
+  cache-cargo-hack:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Fetch latest release version of cargo-hack
+      run: |
+        mkdir -p .github/caching
+        curl -sL https://api.github.com/repos/taiki-e/cargo-hack/releases/latest | jq -r '.name' > .github/caching/cargo-hack.lock
+    - name: Cache cargo-hack/bin
+      id: cache-cargo-hack
+      uses: actions/cache@v1
+      with:
+        path: ${{ runner.tool_cache }}/cargo-hack/bin
+        key: cargo-hack-bin-${{ hashFiles('.github/caching/cargo-hack.lock') }}
+    - name: Install cargo-hack
+      if: "steps.cache-cargo-hack.outputs.cache-hit != 'true'"
+      uses: actions-rs/cargo@v1
+      with:
+        command: install
+        args: --root ${{ runner.tool_cache }}/cargo-hack --force cargo-hack
+
+  cargo-hack:
+    needs: cache-cargo-hack
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@master
+    - uses: actions-rs/toolchain@v1
+      with:
+        toolchain: stable
+        profile: minimal
+    - name: Fetch latest release version of cargo-hack
+      run: |
+        mkdir -p .github/caching
+        curl -sL https://api.github.com/repos/taiki-e/cargo-hack/releases/latest | jq -r '.name' > .github/caching/cargo-hack.lock
+    - name: Restore cargo-hack/bin
+      uses: actions/cache@v1
+      with:
+        path: ${{ runner.tool_cache }}/cargo-hack/bin
+        key: cargo-hack-bin-${{ hashFiles('.github/caching/cargo-hack.lock') }}
+    - run: echo "::add-path::${{ runner.tool_cache }}/cargo-hack/bin"
+    # if `cargo-hack` somehow doesn't exist after loading it from the cache,
+    # make *sure* it's there.
+    - run: cargo hack --help || { cargo install --force cargo-hack; }
+    - name: cargo hack check
+      working-directory: ${{ matrix.subcrate }}
+      run: cargo hack check --each-feature --no-dev-deps --all
+
+  test-versions:
+    # Test against the stable, beta, and nightly Rust toolchains on ubuntu-latest.
+    needs: check
     runs-on: ubuntu-latest
-
     strategy:
       matrix:
-        crate:
-          - tower
-          - tower-layer
-          - tower-service
-          - tower-test
-
+        rust: [stable, beta, nightly, 1.39.0]
     steps:
-      - name: Checkout code
-        uses: actions/checkout@v1
-      - name: Install rust toolchain
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: stable
-          override: true
-      - name: Patch
-        run: |
-          set -e
-          # Remove any existing patch statements
-          mv Cargo.toml Cargo.toml.bck
-          sed -n '/\[patch.crates-io\]/q;p' Cargo.toml.bck > Cargo.toml
-          # Patch all crates
-          cat .github/workflows/patch.toml >> Cargo.toml
-          # Print `Cargo.toml` for debugging
-          echo "~~~~ Cargo.toml ~~~~"
-          cat Cargo.toml
-          echo "~~~~~~~~~~~~~~~~~~~~"
-      - name: Build
-        uses: actions-rs/cargo@v1
-        with:
-          command: check
-          args: -p ${{ matrix.crate }} --verbose
-      - name: Run tests
-        uses: actions-rs/cargo@v1
-        env:
-          CI: 'True'
-          # RUSTFLAGS: '-D warnings'
-        with:
-          command: test
-          args: -p ${{ matrix.crate }} --verbose --all-features
-
-  features:
-
+    - uses: actions/checkout@master
+    - uses: actions-rs/toolchain@v1
+      with:
+        toolchain: ${{ matrix.rust }}
+        profile: minimal
+    - name: Run tests
+      uses: actions-rs/cargo@v1
+      with:
+        command: test
+        args: --all --all-features
+
+  style:
+    # Check style.
+    needs: check
     runs-on: ubuntu-latest
-
     steps:
-      - name: Checkout code
-        uses: actions/checkout@v1
-      - name: Install rust toolchain
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: nightly
-          override: true
-      - name: Patch
-        run: |
-          set -e
-          # Remove any existing patch statements
-          mv Cargo.toml Cargo.toml.bck
-          sed -n '/\[patch.crates-io\]/q;p' Cargo.toml.bck > Cargo.toml
-          # Patch all crates
-          cat .github/workflows/patch.toml >> Cargo.toml
-          # Print `Cargo.toml` for debugging
-          echo "~~~~ Cargo.toml ~~~~"
-          cat Cargo.toml
-          echo "~~~~~~~~~~~~~~~~~~~~"
-      - name: Install cargo-hack
-        uses: actions-rs/cargo@v1
-        with:
-          command: install
-          args: cargo-hack
-      - name: Run feature checks
-        uses: actions-rs/cargo@v1
-        env:
-          CI: 'True'
-        with:
-          command: hack
-          # Check each feature works properly
-          # * --each-feature
-          #   run for each feature which includes --no-default-features and default features of package
-          # * -Z avoid-dev-deps
-          #   build without dev-dependencies to avoid https://github.com/rust-lang/cargo/issues/4866
-          #   tracking-issue: https://github.com/rust-lang/cargo/issues/5133
-          args: check -p tower --each-feature -Z avoid-dev-deps
-
-  rustfmt:
-
+    - uses: actions/checkout@master
+    - uses: actions-rs/toolchain@v1
+      with:
+        toolchain: stable
+        components: rustfmt
+        profile: minimal
+    - name: rustfmt
+      uses: actions-rs/cargo@v1
+      with:
+        command: fmt
+        args: --all -- --check
+  
+  # warnings:
+  #   # Check for any warnings. This is informational and thus is allowed to fail.
+  #   runs-on: ubuntu-latest
+  #   steps:
+  #   - uses: actions/checkout@master
+  #   - uses: actions-rs/toolchain@v1
+  #     with:
+  #       toolchain: stable
+  #       components: clippy
+  #       profile: minimal
+  #   - name: Clippy
+  #     uses: actions-rs/clippy-check@v1
+  #     with:
+  #       token: ${{ secrets.GITHUB_TOKEN }}
+  #       args: --all --all-targets --all-features -- -D warnings
+
+  deny-check:
+    name: cargo-deny check
     runs-on: ubuntu-latest
-
     steps:
-      - name: Checkout code
-        uses: actions/checkout@v1
-      - name: Ensure that rustfmt is installed
-        uses: actions-rs/toolchain@v1
-        with:
-          toolchain: stable
-          override: true
-          components: rustfmt
-      - name: Run rustfmt
-        uses: actions-rs/cargo@v1
-        with:
-          command: fmt
-          args: --all -- --check
-
-# This is failing, because it finds errors...
-#  clippy_check:
-#
-#    runs-on: ubuntu-latest
-#
-#    steps:
-#      - name: Checkout code
-#        uses: actions/checkout@v1
-#      - name: Ensure that clippy is installed
-#        uses: actions-rs/toolchain@v1
-#        with:
-#          toolchain: nightly
-#          override: true
-#          components: clippy
-#      - name: Run clippy
-#        uses: actions-rs/clippy-check@v1
-#        with:
-#          token: ${{ secrets.GITHUB_TOKEN }}
-#          args: --all-targets --all-features
+    - uses: actions/checkout@v1
+    - uses: EmbarkStudios/cargo-deny-action@v0
diff --git a/deny.toml b/deny.toml
new file mode 100644
index 0000000..3284e5e
--- /dev/null
+++ b/deny.toml
@@ -0,0 +1,22 @@
+[advisories]
+vulnerability = "deny"
+unmaintained = "warn"
+notice = "warn"
+ignore = []
+
+[licenses]
+unlicensed = "deny"
+allow = []
+deny = []
+copyleft = "warn"
+allow-osi-fsf-free = "either"
+confidence-threshold = 0.8
+
+[bans]
+multiple-versions = "deny"
+highlight = "all"
+
+[sources]
+unknown-registry = "warn"
+unknown-git = "warn"
+allow-git = []
diff --git a/tower-service/Cargo.toml b/tower-service/Cargo.toml
index 643e16c..155cdc9 100644
--- a/tower-service/Cargo.toml
+++ b/tower-service/Cargo.toml
@@ -24,4 +24,4 @@ edition = "2018"
 [dependencies]
 
 [dev-dependencies]
-http = "0.1"
+http = "0.2"
diff --git a/tower/Cargo.toml b/tower/Cargo.toml
index 2c9c9e5..e15f02e 100644
--- a/tower/Cargo.toml
+++ b/tower/Cargo.toml
@@ -60,10 +60,17 @@ tokio = { version = "0.2", optional = true }
 [dev-dependencies]
 futures-util = { version = "0.3", default-features = false, features = ["alloc", "async-await"] }
 hdrhistogram = "6.0"
-quickcheck = { version = "0.6", default-features = false }
+quickcheck = { version = "0.9", default-features = false }
 tokio = { version = "0.2", features = ["macros", "stream", "sync", "test-util" ] }
 tokio-test = "0.2"
 tower-test = { version = "0.3", path = "../tower-test" }
 tracing-subscriber = "0.1.1"
 # env_logger = { version = "0.5.3", default-features = false }
 # log = "0.4.1"
+
+[package.metadata.docs.rs]
+all-features = true
+rustdoc-args = ["--cfg", "docsrs"]
+
+[package.metadata.playground]
+features = ["full"]
\ No newline at end of file
diff --git a/tower/src/builder/mod.rs b/tower/src/builder/mod.rs
index ca72409..2061443 100644
--- a/tower/src/builder/mod.rs
+++ b/tower/src/builder/mod.rs
@@ -190,6 +190,27 @@ impl<L> ServiceBuilder<L> {
         self.layer(crate::timeout::TimeoutLayer::new(timeout))
     }
 
+    /// Map one request type to another.
+    #[cfg(feature = "util")]
+    pub fn map_request<F, R1, R2>(
+        self,
+        f: F,
+    ) -> ServiceBuilder<Stack<crate::util::MapRequestLayer<F>, L>>
+    where
+        F: Fn(R1) -> R2,
+    {
+        self.layer(crate::util::MapRequestLayer::new(f))
+    }
+
+    /// Map one response type to another.
+    #[cfg(feature = "util")]
+    pub fn map_response<F>(
+        self,
+        f: F,
+    ) -> ServiceBuilder<Stack<crate::util::MapResponseLayer<F>, L>> {
+        self.layer(crate::util::MapResponseLayer::new(f))
+    }
+
     /// Obtains the underlying `Layer` implementation.
     pub fn into_inner(self) -> L {
         self.layer
diff --git a/tower/src/lib.rs b/tower/src/lib.rs
index 7931645..d1e3672 100644
--- a/tower/src/lib.rs
+++ b/tower/src/lib.rs
@@ -48,8 +48,15 @@ pub mod timeout;
 pub mod util;
 
 pub mod builder;
+pub mod layer;
 
 #[cfg(feature = "util")]
+#[doc(inline)]
 pub use self::util::{service_fn, ServiceExt};
+
+#[doc(inline)]
 pub use crate::builder::ServiceBuilder;
+#[doc(inline)]
+pub use tower_layer::Layer;
+#[doc(inline)]
 pub use tower_service::Service;
diff --git a/tower/src/reconnect/future.rs b/tower/src/reconnect/future.rs
index b1d2c74..8ff9a96 100644
--- a/tower/src/reconnect/future.rs
+++ b/tower/src/reconnect/future.rs
@@ -1,5 +1,5 @@
 use super::Error;
-use pin_project::pin_project;
+use pin_project::{pin_project, project};
 use std::{
     future::Future,
     pin::Pin,
@@ -9,25 +9,50 @@ use std::{
 /// Future that resolves to the response or failure to connect.
 #[pin_project]
 #[derive(Debug)]
-pub struct ResponseFuture<F> {
+pub struct ResponseFuture<F, E> {
     #[pin]
-    inner: F,
+    inner: Inner<F, E>,
 }
 
-impl<F> ResponseFuture<F> {
+#[pin_project]
+#[derive(Debug)]
+enum Inner<F, E> {
+    Future(#[pin] F),
+    Error(Option<E>),
+}
+
+impl<F, E> ResponseFuture<F, E> {
     pub(crate) fn new(inner: F) -> Self {
-        ResponseFuture { inner }
+        ResponseFuture {
+            inner: Inner::Future(inner),
+        }
+    }
+
+    pub(crate) fn error(error: E) -> Self {
+        ResponseFuture {
+            inner: Inner::Error(Some(error)),
+        }
     }
 }
 
-impl<F, T, E> Future for ResponseFuture<F>
+impl<F, T, E, ME> Future for ResponseFuture<F, ME>
 where
     F: Future<Output = Result<T, E>>,
     E: Into<Error>,
+    ME: Into<Error>,
 {
     type Output = Result<T, Error>;
 
+    #[project]
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        self.project().inner.poll(cx).map_err(Into::into)
+        let me = self.project();
+        #[project]
+        match me.inner.project() {
+            Inner::Future(fut) => fut.poll(cx).map_err(Into::into),
+            Inner::Error(e) => {
+                let e = e.take().expect("Polled after ready.").into();
+                Poll::Ready(Err(e))
+            }
+        }
     }
 }
diff --git a/tower/src/reconnect/mod.rs b/tower/src/reconnect/mod.rs
index 2464ee4..272579b 100644
--- a/tower/src/reconnect/mod.rs
+++ b/tower/src/reconnect/mod.rs
@@ -1,6 +1,21 @@
-pub mod future;
+//! Reconnect services when they fail.
+//!
+//! Reconnect takes some [`MakeService`] and transforms it into a
+//! [`Service`]. It then attempts to lazily connect and
+//! reconnect on failure. The `Reconnect` service becomes unavailable
+//! when the inner `MakeService::poll_ready` returns an error. When the
+//! connection future returned from `MakeService::call` fails this will be
+//! returned in the next call to `Reconnect::call`. This allows the user to
+//! call the service again even if the inner `MakeService` was unable to
+//! connect on the last call.
+//!
+//! [`MakeService`]: ../make/trait.MakeService.html
+//! [`Service`]: ../trait.Service.html
+
+mod future;
+
+pub use future::ResponseFuture;
 
-use self::future::ResponseFuture;
 use crate::make::MakeService;
 use std::fmt;
 use std::{
@@ -11,6 +26,8 @@ use std::{
 use tower_service::Service;
 use tracing::trace;
 
+pub(crate) type Error = Box<dyn std::error::Error + Send + Sync>;
+
 /// Reconnect to failed services.
 pub struct Reconnect<M, Target>
 where
@@ -19,10 +36,9 @@ where
     mk_service: M,
     state: State<M::Future, M::Response>,
     target: Target,
+    error: Option<M::Error>,
 }
 
-type Error = Box<dyn std::error::Error + Send + Sync>;
-
 #[derive(Debug)]
 enum State<F, S> {
     Idle,
@@ -35,18 +51,12 @@ where
     M: Service<Target>,
 {
     /// Lazily connect and reconnect to a Service.
-    pub fn new<S, Request>(mk_service: M, target: Target) -> Self
-    where
-        M: Service<Target, Response = S>,
-        S: Service<Request>,
-        M::Error: Into<Error>,
-        S::Error: Into<Error>,
-        Target: Clone,
-    {
+    pub fn new<S, Request>(mk_service: M, target: Target) -> Self {
         Reconnect {
             mk_service,
             state: State::Idle,
             target,
+            error: None,
         }
     }
 
@@ -56,6 +66,7 @@ where
             mk_service,
             state: State::Connected(init_conn),
             target,
+            error: None,
         }
     }
 }
@@ -65,24 +76,20 @@ where
     M: Service<Target, Response = S>,
     S: Service<Request>,
     M::Future: Unpin,
-    M::Error: Into<Error>,
-    S::Error: Into<Error>,
+    Error: From<M::Error> + From<S::Error>,
     Target: Clone,
 {
     type Response = S::Response;
     type Error = Error;
-    type Future = ResponseFuture<S::Future>;
+    type Future = ResponseFuture<S::Future, M::Error>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        let ret;
-        let mut state;
-
         loop {
-            match self.state {
+            match &mut self.state {
                 State::Idle => {
                     trace!("poll_ready; idle");
                     match self.mk_service.poll_ready(cx) {
-                        Poll::Ready(r) => r.map_err(Into::into)?,
+                        Poll::Ready(r) => r?,
                         Poll::Pending => {
                             trace!("poll_ready; MakeService not ready");
                             return Poll::Pending;
@@ -97,7 +104,7 @@ where
                     trace!("poll_ready; connecting");
                     match Pin::new(f).poll(cx) {
                         Poll::Ready(Ok(service)) => {
-                            state = State::Connected(service);
+                            self.state = State::Connected(service);
                         }
                         Poll::Pending => {
                             trace!("poll_ready; not ready");
@@ -105,8 +112,8 @@ where
                         }
                         Poll::Ready(Err(e)) => {
                             trace!("poll_ready; error");
-                            state = State::Idle;
-                            ret = Err(e.into());
+                            self.state = State::Idle;
+                            self.error = Some(e.into());
                             break;
                         }
                     }
@@ -124,20 +131,21 @@ where
                         }
                         Poll::Ready(Err(_)) => {
                             trace!("poll_ready; error");
-                            state = State::Idle;
+                            self.state = State::Idle;
                         }
                     }
                 }
             }
-
-            self.state = state;
         }
 
-        self.state = state;
-        Poll::Ready(ret)
+        Poll::Ready(Ok(()))
     }
 
     fn call(&mut self, request: Request) -> Self::Future {
+        if let Some(error) = self.error.take() {
+            return ResponseFuture::error(error);
+        }
+
         let service = match self.state {
             State::Connected(ref mut service) => service,
             _ => panic!("service not ready; poll_ready must be called first"),
diff --git a/tower/src/util/map/mod.rs b/tower/src/util/map/mod.rs
new file mode 100644
index 0000000..2c526d3
--- /dev/null
+++ b/tower/src/util/map/mod.rs
@@ -0,0 +1,7 @@
+mod request;
+mod response;
+
+#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
+pub use request::{MapRequest, MapRequestLayer};
+#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
+pub use response::{MapResponse, MapResponseLayer};
diff --git a/tower/src/util/map/request.rs b/tower/src/util/map/request.rs
new file mode 100644
index 0000000..55562b7
--- /dev/null
+++ b/tower/src/util/map/request.rs
@@ -0,0 +1,70 @@
+use std::task::{Context, Poll};
+use tower_layer::Layer;
+use tower_service::Service;
+
+/// A [`tower::Service`] which maps from one request type to another.
+///
+/// [`tower::Service`]: ../trait.Service.html
+#[derive(Debug, Clone)]
+pub struct MapRequest<S, F> {
+    inner: S,
+    f: F,
+}
+
+impl<S, F> MapRequest<S, F> {
+    /// Create a new `Map` service.
+    pub fn new(inner: S, f: F) -> Self {
+        MapRequest { inner, f }
+    }
+}
+
+impl<S, F, R1, R2> Service<R1> for MapRequest<S, F>
+where
+    S: Service<R2>,
+    F: FnMut(R1) -> R2,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: R1) -> Self::Future {
+        let req = (self.f)(req);
+        self.inner.call(req)
+    }
+}
+
+/// A [`tower::Layer`] that accepts an [`FnMut`] and applies it
+/// to each inner service. Note that [`FnMut`] will implement
+/// [`Clone`] _if_ all the captured values implement either [`Copy`]
+/// or [`Clone`].
+///
+/// [`tower::Layer`]: ../trait.Layer.html
+/// [`FnMut`]: https://doc.rust-lang.org/std/ops/trait.FnMut.html
+/// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html
+/// [`Copy`]: https://doc.rust-lang.org/std/marker/trait.Copy.html
+#[derive(Debug, Clone)]
+pub struct MapRequestLayer<F> {
+    f: F,
+}
+
+impl<F> MapRequestLayer<F> {
+    /// Create a `MapLayer` from some `Fn`.
+    pub fn new(f: F) -> Self {
+        MapRequestLayer { f }
+    }
+}
+
+impl<F, S> Layer<S> for MapRequestLayer<F>
+where
+    F: Clone,
+{
+    type Service = MapRequest<S, F>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        MapRequest::new(inner, self.f.clone())
+    }
+}
diff --git a/tower/src/util/map/response.rs b/tower/src/util/map/response.rs
new file mode 100644
index 0000000..fbe69d0
--- /dev/null
+++ b/tower/src/util/map/response.rs
@@ -0,0 +1,118 @@
+use pin_project::pin_project;
+use std::{
+    future::Future,
+    marker::PhantomData,
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+};
+use tower_layer::Layer;
+use tower_service::Service;
+
+/// A [`tower::Service`] which maps from one response type to another.
+///
+/// [`tower::Service`]: ../trait.Service.html
+#[derive(Debug)]
+pub struct MapResponse<S, F> {
+    inner: S,
+    f: Arc<F>,
+}
+
+impl<S, F> MapResponse<S, F> {
+    /// Create a new `MapResponse`.
+    pub fn new(inner: S, f: F) -> Self {
+        let f = Arc::new(f);
+        MapResponse { inner, f }
+    }
+}
+
+impl<S, F, Request, R1, R2> Service<Request> for MapResponse<S, F>
+where
+    S: Service<Request, Response = R1>,
+    F: Fn(R1) -> R2,
+{
+    type Response = R2;
+    type Error = S::Error;
+    type Future = MapResponseFuture<F, S::Future, R1, R2, S::Error>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: Request) -> Self::Future {
+        let f = self.f.clone();
+        let fut = self.inner.call(req);
+
+        MapResponseFuture {
+            f,
+            fut,
+            _pd: PhantomData,
+        }
+    }
+}
+
+impl<S: Clone, F> Clone for MapResponse<S, F> {
+    fn clone(&self) -> Self {
+        MapResponse {
+            inner: self.inner.clone(),
+            f: self.f.clone(),
+        }
+    }
+}
+
+/// A [`tower::Layer`] that accepts a [`Clone`]able [`Fn`] and applies it
+/// to each inner service.
+///
+/// [`tower::Layer`]: ../trait.Layer.html
+/// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html
+/// [`Fn`]: https://doc.rust-lang.org/std/ops/trait.Fn.html
+#[derive(Debug, Clone)]
+pub struct MapResponseLayer<F> {
+    f: F,
+}
+
+impl<F> MapResponseLayer<F> {
+    /// Create a new `MapResponseLayer`.
+    pub fn new(f: F) -> Self {
+        MapResponseLayer { f }
+    }
+}
+
+impl<S, F> Layer<S> for MapResponseLayer<F>
+where
+    F: Clone,
+{
+    type Service = MapResponse<S, F>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        MapResponse::new(inner, self.f.clone())
+    }
+}
+
+#[pin_project]
+#[derive(Debug)]
+pub struct MapResponseFuture<F, Fut, R1, R2, E> {
+    f: Arc<F>,
+    #[pin]
+    fut: Fut,
+    _pd: PhantomData<fn(R1, R2, E)>,
+}
+
+impl<F, Fut, R1, R2, E> Future for MapResponseFuture<F, Fut, R1, R2, E>
+where
+    F: Fn(R1) -> R2,
+    Fut: Future<Output = Result<R1, E>>,
+{
+    type Output = Result<R2, E>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let me = self.project();
+
+        let res = match futures_core::ready!(me.fut.poll(cx)) {
+            Ok(r) => Ok((&me.f)(r)),
+            Err(e) => Err(e),
+        };
+
+        Poll::Ready(res)
+    }
+}
diff --git a/tower/src/util/mod.rs b/tower/src/util/mod.rs
index 7c642a7..a7b90e7 100644
--- a/tower/src/util/mod.rs
+++ b/tower/src/util/mod.rs
@@ -3,6 +3,7 @@
 mod boxed;
 mod call_all;
 mod either;
+mod map;
 mod oneshot;
 mod optional;
 mod ready;
@@ -11,6 +12,7 @@ mod service_fn;
 pub use self::{
     boxed::{BoxService, UnsyncBoxService},
     either::Either,
+    map::{MapRequest, MapRequestLayer, MapResponse, MapResponseLayer},
     oneshot::Oneshot,
     optional::Optional,
     ready::{Ready, ReadyAnd, ReadyOneshot},


Test patch:
diff --git a/tower/tests/builder.rs b/tower/tests/builder.rs
index a922001..96ffdc7 100644
--- a/tower/tests/builder.rs
+++ b/tower/tests/builder.rs
@@ -2,10 +2,8 @@
 
 use futures_util::{future::Ready, pin_mut};
 use std::time::Duration;
-use tower::buffer::BufferLayer;
 use tower::builder::ServiceBuilder;
-use tower::limit::{concurrency::ConcurrencyLimitLayer, rate::RateLimitLayer};
-use tower::retry::{Policy, RetryLayer};
+use tower::retry::Policy;
 use tower::util::ServiceExt;
 use tower_service::*;
 use tower_test::{assert_request_eq, mock};
@@ -15,41 +13,41 @@ async fn builder_service() {
     let (service, handle) = mock::pair();
     pin_mut!(handle);
 
-    let policy = MockPolicy;
+    let policy = MockPolicy::<&'static str, bool>::default();
     let mut client = ServiceBuilder::new()
-        .layer(BufferLayer::new(5))
-        .layer(ConcurrencyLimitLayer::new(5))
-        .layer(RateLimitLayer::new(5, Duration::from_secs(1)))
-        .layer(RetryLayer::new(policy))
-        .layer(BufferLayer::new(5))
+        .buffer(5)
+        .concurrency_limit(5)
+        .rate_limit(5, Duration::from_secs(5))
+        .retry(policy)
+        .map_response(|r: &'static str| if r == "world" { true } else { false })
+        .map_request(|r: &'static str| if r == "hello" { true } else { false })
         .service(service);
 
     // allow a request through
     handle.allow(1);
 
     let fut = client.ready_and().await.unwrap().call("hello");
-    assert_request_eq!(handle, "hello").send_response("world");
-    assert_eq!(fut.await.unwrap(), "world");
+    assert_request_eq!(handle, true).send_response("world");
+    assert_eq!(fut.await.unwrap(), true);
 }
 
-#[derive(Debug, Clone)]
-struct MockPolicy;
+#[derive(Debug, Clone, Default)]
+struct MockPolicy<Req, Res> {
+    _pd: std::marker::PhantomData<(Req, Res)>,
+}
 
-impl<E> Policy<&'static str, &'static str, E> for MockPolicy
+impl<Req, Res, E> Policy<Req, Res, E> for MockPolicy<Req, Res>
 where
+    Req: Clone,
     E: Into<Box<dyn std::error::Error + Send + Sync + 'static>>,
 {
     type Future = Ready<Self>;
 
-    fn retry(
-        &self,
-        _req: &&'static str,
-        _result: Result<&&'static str, &E>,
-    ) -> Option<Self::Future> {
+    fn retry(&self, _req: &Req, _result: Result<&Res, &E>) -> Option<Self::Future> {
         None
     }
 
-    fn clone_request(&self, req: &&'static str) -> Option<&'static str> {
+    fn clone_request(&self, req: &Req) -> Option<Req> {
         Some(req.clone())
     }
 }