Filter

pub trait AsyncPredicate<Request> {
    // ...
    type Future: Future<Output = Result<Self::Request, BoxError>>;
    // ...
    type Request;
    // ...
    fn check(&mut self, request: Request) -> Self::Future;
}
/// Checks a request synchronously.
pub trait Predicate<Request> {
    // ...
    type Request;
    // ...
    fn check(&mut self, request: Request) -> Result<Self::Request, BoxError>;
}

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx).map_err(Into::into)
    }

    fn call(&mut self, request: Request) -> Self::Future {
        ResponseFuture::new(match self.predicate.check(request) {
            Ok(request) => Either::Right(self.inner.call(request).err_into()),
            Err(e) => Either::Left(std::future::ready(Err(e))),
        })
    }

    fn call(&mut self, request: Request) -> Self::Future {
        use std::mem;

        let inner = self.inner.clone();
        // In case the inner service has state that's driven to readiness and
        // not tracked by clones (such as `Buffer`), pass the version we have
        // already called `poll_ready` on into the future, and leave its clone
        // behind.
        let inner = mem::replace(&mut self.inner, inner);

        // Check the request
        let check = self.predicate.check(request);

        AsyncResponseFuture::new(check, inner)
    }

        loop {
            match this.state.as_mut().project() {
                StateProj::Check { mut check } => {
                    let request = ready!(check.as_mut().poll(cx))?;
                    let response = this.service.call(request);
                    this.state.set(State::WaitResponse { response });
                }
                StateProj::WaitResponse { response } => {
                    return response.poll(cx).map_err(Into::into);
                }
            }
        }

tower = { version = "=0.5.3", features = ["filter", "util"] }
tokio = { version = "=1.54.1", features = ["full", "test-util"] }

use std::num::ParseIntError;
use std::time::Duration;
use tower::{service_fn, BoxError, Service, ServiceBuilder, ServiceExt};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  // Inner service takes u32; the predicate turns a String into one.
  let double = service_fn(|n: u32| async move { Ok::<_, BoxError>(n * 2) });
  let mut svc = ServiceBuilder::new()
    .filter(|s: String| s.trim().parse::<u32>())
    .service(double);

  for input in ["21", "abc"] {
    match svc.ready().await.unwrap().call(input.to_string()).await {
      Ok(v) => println!("filter({input:?}) = {v}"),
      Err(e) => println!(
        "filter({input:?}) rejected: {e} | ParseIntError={}",
        e.downcast_ref::<ParseIntError>().is_some()
      ),
    }
  }

  // The async predicate may await before deciding.
  let echo = service_fn(|s: String| async move { Ok::<_, BoxError>(s) });
  let mut svc = ServiceBuilder::new()
    .filter_async(|user: String| async move {
      tokio::time::sleep(Duration::from_millis(5)).await; // e.g. ask a remote ACL
      if user == "mallory" {
        Err::<String, BoxError>(format!("{user} is banned").into())
      } else {
        Ok(format!("hello, {user}"))
      }
    })
    .service(echo);

  for user in ["alice", "mallory"] {
    let r = svc.ready().await.unwrap().call(user.to_string()).await;
    println!("filter_async({user:?}) = {:?}", r.map_err(|e| e.to_string()));
  }
}

filter("21") = 42
filter("abc") rejected: invalid digit found in string | ParseIntError=true
filter_async("alice") = Ok("hello, alice")
filter_async("mallory") = Err("mallory is banned")

    pub fn layer(predicate: U) -> FilterLayer<U> {
        FilterLayer::new(predicate)
    }

use tower::filter::AsyncFilter;
use tower::{service_fn, BoxError, Layer, Service, ServiceExt};

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let echo = service_fn(|s: String| async move { Ok::<_, BoxError>(s) });
  let layer = AsyncFilter::<(), _>::layer(|s: String| async move { Ok::<_, BoxError>(s) });
  let mut svc = layer.layer(echo);
  let _ = svc.ready().await.unwrap().call("hi".to_string()).await;
}

error[E0599]: the method `ready` exists for struct `Filter<ServiceFn<{closure@...}>, {closure@...}>`, but its trait bounds were not satisfied
  --> examples/async_filter_layer.rs:9:15
   |
 9 |   let _ = svc.ready().await.unwrap().call("hi".to_string()).await;
   |               ^^^^^ method cannot be called on `Filter<ServiceFn<{closure@...}>, {closure@...}>` due to unsatisfied trait bounds