conn::http1 客户端

pub(crate) struct Sender<T, U> {
    /// One message is always allowed, even if the Receiver hasn't asked
    /// for it yet. This boolean keeps track of whether we've sent one
    /// without notice.
    #[cfg(feature = "http1")]
    buffered_once: bool,
    /// The Giver helps watch that the Receiver side has been polled
    /// when the queue is empty. This helps us know when a request and
    /// response have been fully processed, and a connection is ready
    /// for more.
    giver: want::Giver,
    /// Actually bounded by the Giver, plus `buffered_once`.
    inner: mpsc::UnboundedSender<Envelope<T, U>>,
}

    fn can_send(&mut self) -> bool {
        if self.giver.give() || !self.buffered_once {
            // If the receiver is ready *now*, then of course we can send.
            //
            // If the receiver isn't ready yet, but we don't have anything
            // in the channel yet, then allow one message.
            self.buffered_once = true;
            true
        } else {
            false
        }
    }

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

use std::convert::Infallible;
use std::future::poll_fn;
use std::task::Poll;
use std::time::{Duration, Instant};

use bytes::Bytes;
use http_body_util::{Empty, Full};
use hyper::client::conn::http1::SendRequest;
use hyper::{Request, Response};
use hyper_util::rt::TokioIo;
use tokio::net::{TcpListener, TcpStream};

// 只 poll 一次 SendRequest::poll_ready，把结果打印出来
async fn probe(tag: &str, start: Instant, sender: &mut SendRequest<Empty<Bytes>>) {
  let r = poll_fn(|cx| Poll::Ready(sender.poll_ready(cx))).await;
  let shown = match r {
    Poll::Pending => "Pending".to_string(),
    Poll::Ready(Ok(())) => "Ready(Ok)".to_string(),
    Poll::Ready(Err(e)) => format!("Ready(Err({e}))"),
  };
  println!("{:>4}ms {tag:<26} is_ready={:<5} poll_ready={shown}", start.elapsed().as_millis(), sender.is_ready());
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
  let start = Instant::now();
  let listener = TcpListener::bind("127.0.0.1:0").await?;
  let addr = listener.local_addr()?;

  // 服务端：每个请求睡 200ms 再回 "ok"
  tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    let svc = hyper::service::service_fn(|_req| async {
      tokio::time::sleep(Duration::from_millis(200)).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;
  });

  let stream = TcpStream::connect(addr).await?;
  let (mut sender, conn) = hyper::client::conn::http1::handshake(TokioIo::new(stream)).await?;
  let req = |p: &str| Request::get(p).header("host", "x").body(Empty::<Bytes>::new()).unwrap();

  // 1. Connection 还没被 poll 过
  probe("before conn polled", start, &mut sender).await;
  let f1 = sender.send_request(req("/1"));
  let f2 = sender.send_request(req("/2"));
  println!("{:>4}ms send /2 -> {:?}", start.elapsed().as_millis(), f2.await.err());

  // 2. 让 Connection 跑起来；/1 是先前排进去的那一个
  tokio::spawn(conn);
  let t1 = tokio::spawn(async move { f1.await.map(|r| r.status()) });
  tokio::time::sleep(Duration::from_millis(100)).await;
  probe("/1 in flight", start, &mut sender).await;
  let f3 = sender.send_request(req("/3"));
  println!("{:>4}ms send /3 -> {:?}", start.elapsed().as_millis(), f3.await.err());

  let r1 = t1.await?;
  println!("{:>4}ms /1 -> {:?}", start.elapsed().as_millis(), r1);
  probe("after /1 done", start, &mut sender).await;

  // 3. 连接在要请求：连着发两个，第二个被拒
  let f4 = sender.send_request(req("/4"));
  let f5 = sender.send_request(req("/5"));
  println!("{:>4}ms send /5 -> {:?}", start.elapsed().as_millis(), f5.await.err());
  let r4 = f4.await?.status();
  println!("{:>4}ms /4 -> {}", start.elapsed().as_millis(), r4);
  Ok(())
}

   0ms before conn polled         is_ready=false poll_ready=Pending
   0ms send /2 -> Some(hyper::Error(Canceled, "connection was not ready"))
 104ms /1 in flight               is_ready=false poll_ready=Pending
 104ms send /3 -> Some(hyper::Error(Canceled, "connection was not ready"))
 204ms /1 -> Ok(200)
 204ms after /1 done              is_ready=true  poll_ready=Ready(Ok)
 204ms send /5 -> Some(hyper::Error(Canceled, "connection was not ready"))
 407ms /4 -> 200 OK

    pub(crate) fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Option<(T, Callback<T, U>)>> {
        match self.inner.poll_recv(cx) {
            Poll::Ready(item) => {
                // A want signaled after finding the queue empty can land after
                // the Sender's `give()` for the message taken here, as can one
                // signaled on a coop-budget Pending with a message queued.
                // Withdraw it, or the Sender reports this connection as ready
                // while it is still serving this message.
                self.taker.unwant();
                Poll::Ready(item.map(|mut env| env.0.take().expect("envelope not dropped")))
            }
            Poll::Pending => {
                self.taker.want();
                Poll::Pending
            }
        }
    }

            match this.rx.poll_recv(cx) {
                Poll::Ready(Some((req, mut cb))) => {
                    // check that future hasn't been canceled already
                    match cb.poll_canceled(cx) {
                        Poll::Ready(()) => {
                            trace!("request canceled");
                            Poll::Ready(None)
                        }
                        Poll::Pending => {
                            // ... 拆成 RequestHead 和 body
                            this.callback = Some(cb);
                            Poll::Ready(Some(Ok((head, body))))
                        }
                    }
                }
                Poll::Ready(None) => {
                    // user has dropped sender handle
                    trace!("client tx closed");
                    this.rx_closed = true;
                    Poll::Ready(None)
                }
                Poll::Pending => Poll::Pending,
            }

use std::convert::Infallible;

use bytes::Bytes;
use http_body_util::Empty;
use hyper::{Request, Response};
use hyper_util::rt::TokioIo;
use tokio::net::{TcpListener, TcpStream};

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  let server = tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    let svc = hyper::service::service_fn(|req: Request<hyper::body::Incoming>| async move {
      println!("server: got {}", req.uri());
      Ok::<_, Infallible>(Response::new(Empty::<Bytes>::new()))
    });
    hyper::server::conn::http1::Builder::new().serve_connection(TokioIo::new(stream), svc).await
  });
  let stream = TcpStream::connect(addr).await.unwrap();
  let (mut tx, conn) = hyper::client::conn::http1::handshake::<_, Empty<Bytes>>(TokioIo::new(stream)).await.unwrap();
  let conn = tokio::spawn(conn);
  let req = |p: &str| Request::get(p).header("host", "x").body(Empty::new()).unwrap();
  // 请求进了通道，连接任务还没跑，调用方就把 future 扔了
  drop(tx.send_request(req("/never")));
  println!("ready:  {:?}", tx.ready().await);
  println!("client conn task: {:?}", conn.await.unwrap());
  println!("server conn task: {:?}", server.await.unwrap());
}

ready:  Err(hyper::Error(ChannelClosed))
client conn task: Ok(())
server conn task: Ok(())

            match msg {
                Ok((msg, body)) => {
                    if let Some(cb) = self.callback.take() {
                        let res = msg.into_response(body);
                        cb.send(Ok(res));
                        Ok(())
                    } else {
                        // ...
                        Err(crate::Error::new_unexpected_message())
                    }
                }
                Err(err) => {
                    if let Some(cb) = self.callback.take() {
                        cb.send(Err(TrySendError {
                            error: err,
                            message: None,
                        }));
                        Ok(())
                    } else if !self.rx_closed {
                        self.rx.close();
                        if let Some((req, cb)) = self.rx.try_recv() {
                            trace!("canceling queued request with connection error: {}", err);
                            // in this case, the message was never even started, so it's safe to tell
                            // the user that the request was completely canceled
                            cb.send(Err(TrySendError {
                                error: crate::Error::new_canceled().with(err),
                                message: Some(req),
                            }));
                            Ok(())
                        } else {
                            Err(err)
                        }
                    } else {
                        Err(err)
                    }
                }
            }

pub(crate) enum Callback<T, U> {
    #[allow(unused)]
    Retry(Option<oneshot::Sender<Result<U, TrySendError<T>>>>),
    NoRetry(Option<oneshot::Sender<Result<U, crate::Error>>>),
}

impl<T, U> Drop for Envelope<T, U> {
    fn drop(&mut self) {
        if let Some((val, cb)) = self.0.take() {
            cb.send(Err(TrySendError {
                error: crate::Error::new_canceled().with("connection closed"),
                message: Some(val),
            }));
        }
    }
}

#[cold]
fn dispatch_gone() -> crate::Error {
    // ...
    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 = ["tokio"] }
http-body-util = "=0.1.5"
bytes = "1"
tokio = { version = "=1.54.1", features = ["macros", "rt", "rt-multi-thread", "net", "time"] }

use std::convert::Infallible;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

use bytes::Bytes;
use http_body_util::Empty;
use hyper::body::{Body, Frame};
use hyper::{Request, Response};
use hyper_util::rt::TokioIo;
use tokio::net::{TcpListener, TcpStream};

// 一个 poll 就 panic 的请求 body
struct PanicBody;
impl Body for PanicBody {
  type Data = Bytes;
  type Error = Infallible;
  fn poll_frame(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Result<Frame<Bytes>, Infallible>>> {
    panic!("body exploded");
  }
}

async fn connect<B>(addr: std::net::SocketAddr) -> (hyper::client::conn::http1::SendRequest<B>, hyper::client::conn::http1::Connection<TokioIo<TcpStream>, B>)
where
  B: Body + 'static + Send,
  B::Data: Send,
  B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
{
  let stream = TcpStream::connect(addr).await.unwrap();
  hyper::client::conn::http1::handshake(TokioIo::new(stream)).await.unwrap()
}

fn get() -> Request<Empty<Bytes>> {
  Request::get("/").header("host", "x").body(Empty::new()).unwrap()
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  std::panic::set_hook(Box::new(|_| {})); // 不打印 panic 信息
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  // 服务端：每个请求睡 200ms 再回
  tokio::spawn(async move {
    loop {
      let (stream, _) = listener.accept().await.unwrap();
      tokio::spawn(async move {
        let svc = hyper::service::service_fn(|_req| async {
          tokio::time::sleep(Duration::from_millis(200)).await;
          Ok::<_, Infallible>(Response::new(Empty::<Bytes>::new()))
        });
        let _ = hyper::server::conn::http1::Builder::new().serve_connection(TokioIo::new(stream), svc).await;
      });
    }
  });

  // A: 请求已写出、在等响应时，abort 连接任务
  let (mut tx, conn) = connect::<Empty<Bytes>>(addr).await;
  let task = tokio::spawn(conn);
  let fut = tx.send_request(get());
  let waiter = tokio::spawn(fut);
  tokio::time::sleep(Duration::from_millis(50)).await;
  task.abort();
  println!("A abort in flight:   {:?}", waiter.await.unwrap().err());

  // B: 连接任务跑在另一个运行时上，请求在途时那个运行时被关掉
  let other = tokio::runtime::Builder::new_multi_thread().worker_threads(1).enable_all().build().unwrap();
  let (mut tx, conn) = connect::<Empty<Bytes>>(addr).await;
  other.spawn(conn);
  let waiter = tokio::spawn(tx.send_request(get()));
  tokio::time::sleep(Duration::from_millis(50)).await;
  other.shutdown_background();
  println!("B runtime shut down: {:?}", waiter.await.unwrap().err());

  // C: 请求 body 在连接任务里 panic
  let (mut tx, conn) = connect::<PanicBody>(addr).await;
  let task = tokio::spawn(conn);
  let res = tx.send_request(Request::post("/").header("host", "x").body(PanicBody).unwrap()).await;
  println!("C body panicked:     {:?}", res.err());
  println!("  conn task:         panic = {}", task.await.unwrap_err().is_panic());

  // D: 请求还在通道里（连接任务一次都没跑过）就 abort
  let (mut tx, conn) = connect::<Empty<Bytes>>(addr).await;
  let task = tokio::spawn(conn);
  let fut = tx.send_request(get());
  task.abort();
  println!("D abort before run:  {:?}", fut.await.err());
}

A abort in flight:   Some(hyper::Error(User(DispatchGone), "runtime dropped the dispatch task"))
B runtime shut down: Some(hyper::Error(User(DispatchGone), "runtime dropped the dispatch task"))
C body panicked:     Some(hyper::Error(User(DispatchGone), "user code panicked"))
  conn task:         panic = true
D abort before run:  Some(hyper::Error(Canceled, "connection closed"))