dispatch task is gone

    crate::Error::new_user_dispatch_gone().with(if std::thread::panicking() {
        "user code panicked"
    } else {
        "runtime dropped the dispatch task"
    })

hyper = { version = "=1.12.0", features = ["client", "server", "http1"] }
hyper-util = { version = "=0.1.21", features = ["client-legacy", "http1", "tokio"] }
http-body-util = "=0.1.5"
bytes = "1"
tokio = { version = "=1.54.1", features = ["macros", "rt-multi-thread", "net", "time", "io-util"] }

// tests/shared_client.rs
use std::convert::Infallible;
use std::net::SocketAddr;
use std::sync::LazyLock;
use std::time::Duration;

use bytes::Bytes;
use http_body_util::{BodyExt, Empty, Full};
use hyper::{Request, Response};
use hyper_util::client::legacy::Client;
use hyper_util::client::legacy::connect::HttpConnector;
use hyper_util::rt::{TokioExecutor, TokioIo};

// A test server on its own thread and runtime; every response takes 20ms.
static SERVER: LazyLock<SocketAddr> = LazyLock::new(|| {
  let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
  let addr = listener.local_addr().unwrap();
  listener.set_nonblocking(true).unwrap();
  std::thread::spawn(move || {
    let rt = tokio::runtime::Runtime::new().unwrap();
    rt.block_on(async move {
      let listener = tokio::net::TcpListener::from_std(listener).unwrap();
      loop {
        let (stream, _) = listener.accept().await.unwrap();
        tokio::spawn(async move {
          let svc = hyper::service::service_fn(|_req: Request<hyper::body::Incoming>| async {
            tokio::time::sleep(Duration::from_millis(20)).await;
            Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"ok"))))
          });
          let _ = hyper::server::conn::http1::Builder::new()
            .serve_connection(TokioIo::new(stream), svc)
            .await;
        });
      }
    });
  });
  addr
});

// One client for the whole test binary.
static CLIENT: LazyLock<Client<HttpConnector, Empty<Bytes>>> =
  LazyLock::new(|| Client::builder(TokioExecutor::new()).build_http());

async fn get_ok(n: usize) {
  let url: hyper::Uri = format!("http://{}/", *SERVER).parse().unwrap();
  for _ in 0..n {
    let resp = CLIENT.get(url.clone()).await.unwrap();
    resp.into_body().collect().await.unwrap();
  }
}

#[tokio::test]
async fn a() {
  get_ok(3).await;
}

#[tokio::test]
async fn b() {
  get_ok(3).await;
}

#[tokio::test]
async fn c() {
  get_ok(3).await;
}

#[tokio::test]
async fn d() {
  get_ok(3).await;
}

running 4 tests
. 1/4
c --- FAILED
a --- FAILED
b --- FAILED

failures:

---- c stdout ----

thread 'c' (780646) panicked at tests/shared_client.rs:46:46:
called `Result::unwrap()` on an `Err` value: hyper_util::client::legacy::Error(SendRequest, hyper::Error(User(DispatchGone), "runtime dropped the dispatch task"))
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

---- a stdout ----

thread 'a' (780644) panicked at tests/shared_client.rs:46:46:
called `Result::unwrap()` on an `Err` value: hyper_util::client::legacy::Error(SendRequest, hyper::Error(User(DispatchGone), "user code panicked"))

---- b stdout ----

thread 'b' (780645) panicked at tests/shared_client.rs:46:46:
called `Result::unwrap()` on an `Err` value: hyper_util::client::legacy::Error(SendRequest, hyper::Error(User(DispatchGone), "user code panicked"))


failures:
    a
    b
    c

test result: FAILED. 1 passed; 3 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.07s

// tests/shared_client_exec.rs：只有 CLIENT 不同
// The connection tasks run on a runtime that lives as long as the test binary.
static RT: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
  tokio::runtime::Builder::new_multi_thread().worker_threads(1).enable_all().build().unwrap()
});

#[derive(Clone)]
struct LongLived;

impl<F> hyper::rt::Executor<F> for LongLived
where
  F: Future + Send + 'static,
  F::Output: Send + 'static,
{
  fn execute(&self, fut: F) {
    RT.spawn(fut);
  }
}

static CLIENT: LazyLock<Client<HttpConnector, Empty<Bytes>>> =
  LazyLock::new(|| Client::builder(LongLived).build_http());

called `Result::unwrap()` on an `Err` value: hyper_util::client::legacy::Error(SendRequest, hyper::Error(Io, Custom { kind: Other, error: "A Tokio 1.x context was found, but it is being shutdown." }))

// tests/shared_client_rt.rs：只有 CLIENT 和 get_ok 不同
// The client, its sockets and its connection tasks all live on this runtime.
static RT: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
  tokio::runtime::Builder::new_multi_thread().worker_threads(1).enable_all().build().unwrap()
});

static CLIENT: LazyLock<Client<HttpConnector, Empty<Bytes>>> =
  LazyLock::new(|| Client::builder(TokioExecutor::new()).build_http());

async fn get_ok(n: usize) {
  let url: hyper::Uri = format!("http://{}/", *SERVER).parse().unwrap();
  // Run the requests on RT; the test's own runtime only waits for the result.
  RT.spawn(async move {
    for _ in 0..n {
      let resp = CLIENT.get(url.clone()).await.unwrap();
      resp.into_body().collect().await.unwrap();
    }
  })
  .await
  .unwrap();
}

use std::error::Error as _;
use std::time::Duration;

use bytes::Bytes;
use http_body_util::{BodyExt, Empty};
use hyper::Request;
use hyper::client::conn::http1;
use hyper_util::rt::TokioIo;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

#[tokio::main(flavor = "current_thread")]
async fn main() {
  // Server: response head plus 3 of 10 body bytes, then nothing for a while.
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move {
    let (mut s, _) = listener.accept().await.unwrap();
    let mut buf = [0u8; 1024];
    let _ = s.read(&mut buf).await.unwrap();
    s.write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 10\r\n\r\nabc").await.unwrap();
    tokio::time::sleep(Duration::from_secs(5)).await;
  });

  // The connection task runs on another runtime, which is shut down mid-body.
  let other = tokio::runtime::Builder::new_multi_thread().worker_threads(1).enable_all().build().unwrap();
  let io = TokioIo::new(TcpStream::connect(addr).await.unwrap());
  let (mut tx, conn) = http1::handshake::<_, Empty<Bytes>>(io).await.unwrap();
  other.spawn(conn);
  let resp = tx.send_request(Request::get("/").header("host", "x").body(Empty::new()).unwrap()).await.unwrap();
  println!("head: {}", resp.status());
  let mut body = resp.into_body();
  let first = body.frame().await.unwrap().unwrap().into_data().unwrap();
  println!("first frame: {first:?}");
  other.shutdown_background();
  match body.frame().await {
    Some(Err(e)) => println!("next frame: Err({e}) source={:?} incomplete={}", e.source().map(|s| s.to_string()), e.is_incomplete_message()),
    other => println!("next frame: {other:?}"),
  }
}

head: 200 OK
first frame: b"abc"
next frame: Err(connection closed before message completed) source=None incomplete=true