在途请求数只增不减

tower-http = { version = "=0.7.1", features = ["metrics"] }
axum = { version = "=0.8.9", default-features = false, features = ["tokio", "http1"] }
futures = "0.3"
tokio = { version = "=1.54.1", features = ["macros", "rt", "net", "time", "io-util"] }

use axum::body::{Body, Bytes};
use axum::routing::get;
use axum::Router;
use futures::stream::{self, StreamExt};
use std::convert::Infallible;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::time::{sleep, Instant};
use tower_http::metrics::InFlightRequestsLayer;

// 每 200 ms 一个事件，永不结束
fn ticking() -> Body {
  let s = stream::unfold(0u32, |n| async move {
    if n > 0 {
      sleep(Duration::from_millis(200)).await;
    }
    Some((Ok::<_, Infallible>(Bytes::from(format!("data: tick {n}\n\n"))), n + 1))
  });
  Body::from_stream(s)
}

// 一个事件之后再也不发
fn silent() -> Body {
  let first = stream::once(async { Ok::<_, Infallible>(Bytes::from("data: hello\n\n")) });
  Body::from_stream(first.chain(stream::pending()))
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let (layer, counter) = InFlightRequestsLayer::pair();
  let app = Router::new()
    .route("/ticking", get(|| async { ticking() }))
    .route("/silent", get(|| async { silent() }))
    .layer(layer);
  let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });

  let start = Instant::now();
  let t = move || start.elapsed().as_millis() / 50 * 50;
  let open = |path: &'static str| async move {
    let mut sock = TcpStream::connect(addr).await.unwrap();
    sock.write_all(format!("GET {path} HTTP/1.1\r\nhost: x\r\n\r\n").as_bytes()).await.unwrap();
    let mut buf = [0u8; 1024];
    let _ = sock.read(&mut buf).await.unwrap();
    sock
  };
  // 三个标签页各开一条事件流；客户端读到第一个事件后就不再读
  let mut socks = vec![open("/ticking").await, open("/silent").await, open("/silent").await];
  sleep(Duration::from_millis(50)).await;
  println!("{:>5}ms three streams open, in flight = {}", t(), counter.get());
  sleep(Duration::from_millis(1000)).await;
  println!("{:>5}ms still open, in flight = {}", t(), counter.get());
  while let Some(sock) = socks.pop() {
    drop(sock);
    sleep(Duration::from_millis(50)).await;
    println!("{:>5}ms one client closed its socket, in flight = {}", t(), counter.get());
  }
}

   50ms three streams open, in flight = 3
 1050ms still open, in flight = 3
 1100ms one client closed its socket, in flight = 2
 1150ms one client closed its socket, in flight = 1
 1200ms one client closed its socket, in flight = 0