poll_ready must be called first：没就绪就 call

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

use std::panic::{AssertUnwindSafe, catch_unwind};
use std::time::Duration;
use tower::buffer::Buffer;
use tower::limit::{ConcurrencyLimit, RateLimit, rate::Rate};
use tower::reconnect::Reconnect;
use tower::{BoxError, Service, service_fn};

fn echo() -> impl Service<u32, Response = u32, Error = BoxError, Future: Send> + Send + 'static {
  service_fn(|x: u32| async move { Ok::<_, BoxError>(x) })
}

// Call `f` and print the panic message (or "no panic").
fn try_call(name: &str, f: impl FnOnce()) {
  match catch_unwind(AssertUnwindSafe(f)) {
    Ok(()) => println!("{name}: no panic"),
    Err(p) => {
      let msg = p
        .downcast_ref::<&str>()
        .map(|s| s.to_string())
        .or_else(|| p.downcast_ref::<String>().cloned())
        .unwrap();
      println!("{name}: {msg}");
    }
  }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let mut cl = ConcurrencyLimit::new(echo(), 4);
  try_call("ConcurrencyLimit", || drop(cl.call(1)));

  let mut rl = RateLimit::new(echo(), Rate::new(1, Duration::from_secs(1)));
  try_call("RateLimit, 1st call", || drop(rl.call(1)));
  try_call("RateLimit, 2nd call", || drop(rl.call(2)));

  let mk = service_fn(|_: ()| std::future::ready(Ok::<_, BoxError>(echo())));
  let mut rc = Reconnect::new(mk, ());
  try_call("Reconnect", || drop(Service::<u32>::call(&mut rc, 1)));

  let mut bf = Buffer::new(echo(), 4);
  try_call("Buffer", || drop(bf.call(1)));
}

thread 'main' (124641) 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
ConcurrencyLimit: max requests in-flight; poll_ready must be called first
RateLimit, 1st call: no panic

thread 'main' (124641) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tower-0.5.3/src/limit/rate/service.rs:115:31:
service not ready; poll_ready must be called first
RateLimit, 2nd call: service not ready; poll_ready must be called first

thread 'main' (124641) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tower-0.5.3/src/reconnect/mod.rs:149:18:
service not ready; poll_ready must be called first
Reconnect: service not ready; poll_ready must be called first

thread 'main' (124641) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tower-0.5.3/src/buffer/service.rs:122:23:
`send_item` called without first calling `poll_reserve`
Buffer: `send_item` called without first calling `poll_reserve`

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

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use tower::{BoxError, Service, ServiceBuilder, ServiceExt, service_fn};

#[derive(Clone, Copy, Debug)]
enum Mode {
  CloneThenCall,
  CloneThenOneshot,
  Replace,
}

// A hand-written middleware whose call moves the inner service into a boxed future.
#[derive(Clone)]
struct Log<S> {
  inner: S,
  mode: Mode,
}

impl<S> Service<u32> for Log<S>
where
  S: Service<u32, Response = u32, Error = BoxError> + Clone + Send + 'static,
  S::Future: Send,
{
  type Response = u32;
  type Error = BoxError;
  type Future = Pin<Box<dyn Future<Output = Result<u32, BoxError>> + Send>>;

  fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), BoxError>> {
    self.inner.poll_ready(cx)
  }

  fn call(&mut self, req: u32) -> Self::Future {
    match self.mode {
      Mode::CloneThenCall => {
        let mut inner = self.inner.clone();
        Box::pin(async move { inner.call(req).await })
      }
      Mode::CloneThenOneshot => {
        let inner = self.inner.clone();
        Box::pin(async move { inner.oneshot(req).await })
      }
      Mode::Replace => {
        let clone = self.inner.clone();
        let mut inner = std::mem::replace(&mut self.inner, clone);
        Box::pin(async move { inner.call(req).await })
      }
    }
  }
}

async fn run<S>(label: &str, inner: S, mode: Mode)
where
  S: Service<u32, Response = u32, Error = BoxError> + Clone + Send + 'static,
  S::Future: Send,
{
  let mut svc = Log { inner, mode };
  let task = tokio::spawn(async move {
    let mut out = Vec::new();
    for i in 1..=2 {
      out.push(svc.ready().await?.call(i).await?);
    }
    Ok::<_, BoxError>(out)
  });
  let result = match tokio::time::timeout(Duration::from_secs(5), task).await {
    Err(_) => "still waiting after 5s".to_string(),
    Ok(Ok(r)) => format!("{:?}", r.map_err(|e| e.to_string())),
    Ok(Err(e)) => {
      let p = e.into_panic();
      let msg = p
        .downcast_ref::<&str>()
        .map(|s| s.to_string())
        .or_else(|| p.downcast_ref::<String>().cloned())
        .unwrap();
      format!("panic: {msg}")
    }
  };
  println!("{label:<10} {:<17} -> {result}", format!("{mode:?}"));
}

fn echo() -> impl Service<u32, Response = u32, Error = BoxError, Future: Send> + Clone + Send + 'static {
  service_fn(|x: u32| async move { Ok::<_, BoxError>(x) })
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  std::panic::set_hook(Box::new(|_| {})); // keep the default hook quiet
  for mode in [Mode::CloneThenCall, Mode::CloneThenOneshot, Mode::Replace] {
    run("limit(1)", ServiceBuilder::new().concurrency_limit(1).service(echo()), mode).await;
    run("buffer(1)", ServiceBuilder::new().buffer(1).service(echo()), mode).await;
  }
}

limit(1)   CloneThenCall     -> panic: max requests in-flight; poll_ready must be called first
buffer(1)  CloneThenCall     -> panic: `send_item` called without first calling `poll_reserve`
limit(1)   CloneThenOneshot  -> still waiting after 5s
buffer(1)  CloneThenOneshot  -> still waiting after 5s
limit(1)   Replace           -> Ok([1, 2])
buffer(1)  Replace           -> Ok([1, 2])