Service trait

pub trait Service<Request> {
    type Response;
    type Error;
    type Future: Future<Output = Result<Self::Response, Self::Error>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;

    #[must_use = "futures do nothing unless you `.await` or poll them"]
    fn call(&mut self, req: Request) -> Self::Future;
}

    /// Returns `Poll::Ready(Ok(()))` when the service is able to process requests.
    ///
    /// If the service is at capacity, then `Poll::Pending` is returned and the task
    /// is notified when the service becomes ready again. This function is
    /// expected to be called while on a task. Generally, this can be done with
    /// a simple [`std::future::poll_fn`] call.
    ///
    /// If `Poll::Ready(Err(_))` is returned, the service is no longer able to service requests
    /// and the caller should discard the service instance.
    ///
    /// Once `poll_ready` returns `Poll::Ready(Ok(()))`, a request may be dispatched to the
    /// service using `call`. Until a request is dispatched, repeated calls to
    /// `poll_ready` must return either `Poll::Ready(Ok(()))` or `Poll::Ready(Err(_))`.
    ///
    /// Note that `poll_ready` may reserve shared resources that are consumed in a subsequent
    /// invocation of `call`. Thus, it is critical for implementations to not assume that `call`
    /// will always be invoked and to ensure that such resources are released if the service is
    /// dropped before `call` is invoked or the future returned by `call` is dropped before it
    /// is polled.
    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;

    /// Process the request and return the response asynchronously.
    ///
    /// This function is expected to be callable off task. As such,
    /// implementations should take care to not call `poll_ready`.
    ///
    /// Before dispatching a request, `poll_ready` must be called and return
    /// `Poll::Ready(Ok(()))`.
    ///
    /// # Panics
    ///
    /// Implementations are permitted to panic if `call` is invoked without
    /// obtaining `Poll::Ready(Ok(()))` from `poll_ready`.

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        // If we haven't already acquired a permit from the semaphore, try to
        // acquire one first.
        if self.permit.is_none() {
            self.permit = ready!(self.semaphore.poll_acquire(cx));
            // ...
        }

        // Once we've acquired a permit (or if we already had one), poll the
        // inner service.
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, request: Request) -> Self::Future {
        // Take the permit
        let permit = self
            .permit
            .take()
            .expect("max requests in-flight; poll_ready must be called first");

        // Call the inner service
        let future = self.inner.call(request);

        ResponseFuture::new(future, permit)
    }

impl<T: Clone> Clone for ConcurrencyLimit<T> {
    fn clone(&self) -> Self {
        // Since we hold an `OwnedSemaphorePermit`, we can't derive `Clone`.
        // Instead, when cloning the service, create a new service with the
        // same semaphore, but with the permit in the un-acquired state.
        Self {
            inner: self.inner.clone(),
            semaphore: self.semaphore.clone(),
            permit: None,
        }
    }
}

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

use std::convert::Infallible;
use std::panic;
use std::sync::Arc;
use tokio::sync::Semaphore;
use tokio_test::task;
use tower::limit::ConcurrencyLimit;
use tower::{service_fn, Service};

fn main() {
  let sem = Arc::new(Semaphore::new(1));
  let echo = service_fn(|req: u32| async move { Ok::<_, Infallible>(req) });
  let mut a = ConcurrencyLimit::with_semaphore(echo, sem.clone());

  let mut ta = task::spawn(());
  let r1 = ta.enter(|cx, _| a.poll_ready(cx));
  let r2 = ta.enter(|cx, _| a.poll_ready(cx));
  println!("a: {:?} then {:?}, permits left = {}", r1, r2, sem.available_permits());

  let mut b = a.clone();
  let mut tb = task::spawn(());
  let rb = tb.enter(|cx, _| b.poll_ready(cx));
  println!("b = a.clone(): {:?}", rb);

  let caught = panic::catch_unwind(panic::AssertUnwindSafe(|| {
    let _ = b.call(1);
  }));
  let msg = caught.unwrap_err();
  println!("b.call panicked: {:?}", msg.downcast_ref::<String>().unwrap());

  drop(a);
  println!("drop(a) without call: permits left = {}, b woken = {}", sem.available_permits(), tb.is_woken());
  let rb = tb.enter(|cx, _| b.poll_ready(cx));
  println!("b again: {:?}, permits left = {}", rb, sem.available_permits());
}

a: Ready(Ok(())) then Ready(Ok(())), permits left = 0
b = a.clone(): Pending

thread 'main' (16795896) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tower-0.5.3/src/limit/concurrency/service.rs:86:14:
max requests in-flight; poll_ready must be called first
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
b.call panicked: "max requests in-flight; poll_ready must be called first"
drop(a) without call: permits left = 0, b woken = true
b again: Ready(Ok(())), permits left = 0

/// Services are permitted to panic if `call` is invoked without obtaining `Poll::Ready(Ok(()))`
/// from `poll_ready`. You should therefore be careful when cloning services for example to move
/// them into boxed futures. Even though the original service is ready, the clone might not be.

    fn call(&mut self, req: R) -> Self::Future {
        let clone = self.inner.clone();
        // take the service that was ready
        let mut inner = std::mem::replace(&mut self.inner, clone);
        Box::pin(async move {
            inner.call(req).await
        })
    }

impl<S, Request> Service<Request> for &mut S
where
    S: Service<Request> + ?Sized,
{
    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).poll_ready(cx)
    }

    fn call(&mut self, request: Request) -> S::Future {
        (**self).call(request)
    }
}

impl<S, Request> Service<Request> for Box<S>
where
    S: Service<Request> + ?Sized,
{
    // 同上
}

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

use std::convert::Infallible;
use std::future::{ready, Ready};
use tower::{service_fn, Service, ServiceExt};

type Fut = Ready<Result<u32, Infallible>>;

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let mut svc = service_fn(|req: u32| ready(Ok::<_, Infallible>(req * 10)));

  // &mut S is a Service: oneshot consumes the reference, not svc
  let r = (&mut svc).oneshot(1).await.unwrap();
  println!("via &mut: {r}");
  let r = svc.ready().await.unwrap().call(2).await.unwrap();
  println!("svc still usable: {r}");

  // Box<dyn Service> is a Service, because Box<S> allows S: ?Sized
  let mut boxed: Box<dyn Service<u32, Response = u32, Error = Infallible, Future = Fut>> =
    Box::new(svc);
  let r = boxed.ready().await.unwrap().call(3).await.unwrap();
  println!("via Box<dyn>: {r}");
}

via &mut: 10
svc still usable: 20
via Box<dyn>: 30

type DynSvc = dyn Service<u32, Response = u32, Error = Infallible, Future = Ready<Result<u32, Infallible>>>;

fn needs_service<S: Service<u32>>(_svc: S) {}

fn pass_dyn(boxed: Box<DynSvc>, by_ref: &mut DynSvc) {
  needs_service(boxed); // Box<S>: S may be unsized
  needs_service(by_ref); // &mut S: S must be Sized in tower-service 0.3.3
}

error[E0277]: the trait bound `&mut _: Service<u32>` is not satisfied
  --> examples/mutdyn.rs:11:17
   |
11 |   needs_service(by_ref); // &mut S: S must be Sized in tower-service 0.3.3
   |   ------------- ^^^^^^ the trait `Sized` is not implemented for `dyn Service<u32, Error = Infallible, Future = _, Response = u32>`
   |   |
   |   required by a bound introduced by this call

after ready: 0
after call:  0
after drop(svc): 0
after drop(fut): 1