auto：一个端口两个协议

pub struct Builder<E> {
    #[cfg(feature = "http1")]
    http1: http1::Builder,
    #[cfg(feature = "http2")]
    http2: http2::Builder<E>,
    #[cfg(any(feature = "http1", feature = "http2"))]
    version: Option<Version>,
    #[cfg(not(feature = "http2"))]
    _executor: E,
}

const H2_PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        if *this.cancelled {
            return Poll::Ready(Err(io::Error::new(io::ErrorKind::Interrupted, "Cancelled")));
        }

        let mut buf = ReadBuf::uninit(&mut *this.buf);
        // SAFETY: `this.filled` tracks how many bytes have been read (and thus initialized) and
        // we're only advancing by that many.
        unsafe {
            buf.unfilled().advance(*this.filled);
        };

        // We start as H2 and switch to H1 as soon as we don't have the preface.
        while buf.filled().len() < H2_PREFACE.len() {
            let len = buf.filled().len();
            ready!(Pin::new(this.io.as_mut().unwrap()).poll_read(cx, buf.unfilled()))?;
            *this.filled = buf.filled().len();

            // We starts as H2 and switch to H1 when we don't get the preface.
            if buf.filled().len() == len
                || buf.filled()[len..] != H2_PREFACE[len..buf.filled().len()]
            {
                *this.version = Version::H1;
                break;
            }
        }

        let io = this.io.take().unwrap();
        let buf = buf.filled().to_vec();
        Poll::Ready(Ok((
            *this.version,
            Rewind::new_buffered(io, Bytes::from(buf)),
        )))
    }

        if let Some(mut prefix) = self.pre.take() {
            // If there are no remaining bytes, let the bytes get dropped.
            if !prefix.is_empty() {
                let copy_len = cmp::min(prefix.len(), buf.remaining());
                buf.put_slice(&prefix[..copy_len]);
                prefix.advance(copy_len);
                // Put back what's left
                if !prefix.is_empty() {
                    self.pre = Some(prefix);
                }

                return Poll::Ready(Ok(()));
            }
        }
        Pin::new(&mut self.inner).poll_read(cx, buf)

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

use std::convert::Infallible;

use bytes::Bytes;
use http_body_util::{BodyExt, Empty, Full};
use hyper::body::Incoming;
use hyper::client::conn::{http1 as client_h1, http2 as client_h2};
use hyper::service::service_fn;
use hyper::{Request, Response};
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

async fn handler(req: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
  let text = format!("{:?} {}", req.version(), req.uri().path());
  Ok(Response::new(Full::new(Bytes::from(text))))
}

#[tokio::main]
async fn main() {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move {
    loop {
      let (stream, _) = listener.accept().await.unwrap();
      tokio::spawn(async move {
        let builder = auto::Builder::new(TokioExecutor::new());
        let r = builder.serve_connection(TokioIo::new(stream), service_fn(handler)).await;
        println!("  server conn ended: {r:?}");
      });
    }
  });

  // 1. hyper HTTP/1 client
  let io = TokioIo::new(TcpStream::connect(addr).await.unwrap());
  let (mut tx, conn) = client_h1::handshake::<_, Empty<Bytes>>(io).await.unwrap();
  let conn_task = tokio::spawn(conn);
  let res = tx.send_request(Request::get("/one").body(Empty::new()).unwrap()).await.unwrap();
  let body = res.into_body().collect().await.unwrap().to_bytes();
  println!("h1 client  -> {:?}", body);
  drop(tx);
  conn_task.await.unwrap().unwrap();

  // 2. hyper HTTP/2 client, prior knowledge (no TLS, no ALPN, no Upgrade)
  let io = TokioIo::new(TcpStream::connect(addr).await.unwrap());
  let (mut tx, conn) = client_h2::handshake::<_, _, Empty<Bytes>>(TokioExecutor::new(), io).await.unwrap();
  let conn_task = tokio::spawn(conn);
  let res = tx
    .send_request(Request::get("http://x/two").body(Empty::new()).unwrap())
    .await
    .unwrap();
  let body = res.into_body().collect().await.unwrap().to_bytes();
  println!("h2 client  -> {:?}", body);
  drop(tx);
  conn_task.await.unwrap().unwrap();

  // 3. raw bytes: an HTTP/1.1 request whose method happens to be PRI
  let mut tcp = TcpStream::connect(addr).await.unwrap();
  tcp.write_all(b"PRI /three HTTP/1.1\r\nconnection: close\r\n\r\n").await.unwrap();
  let mut out = String::new();
  tcp.read_to_string(&mut out).await.unwrap();
  println!("raw PRI    -> {:?}", out.lines().next().unwrap_or(""));
  println!("              body {:?}", out.rsplit("\r\n").next().unwrap_or(""));
  tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}

h1 client  -> b"HTTP/1.1 /one"
  server conn ended: Ok(())
h2 client  -> b"HTTP/2.0 /two"
  server conn ended: Ok(())
  server conn ended: Ok(())
raw PRI    -> "HTTP/1.1 200 OK"
              body "HTTP/1.1 /three"

    enum ConnState<'a, I, S, E>
    where
        S: HttpService<Incoming>,
    {
        ReadVersion {
            #[pin]
            read_version: ReadVersion<I>,
            builder: Cow<'a, Builder<E>>,
            service: Option<S>,
        },
        H1 {
            #[pin]
            conn: Http1Connection<I, S>,
        },
        H2 {
            #[pin]
            conn: Http2Connection<I, S, E>,
        },
    }

        let state = match self.version {
            #[cfg(feature = "http1")]
            Some(Version::H1) => {
                let io = Rewind::new_buffered(io, Bytes::new());
                let conn = self.http1.serve_connection(io, service);
                ConnState::H1 { conn }
            }
            #[cfg(feature = "http2")]
            Some(Version::H2) => {
                let io = Rewind::new_buffered(io, Bytes::new());
                let conn = self.http2.serve_connection(io, service);
                ConnState::H2 { conn }
            }
            #[cfg(any(feature = "http1", feature = "http2"))]
            _ => ConnState::ReadVersion {
                read_version: read_version(io),
                builder: Cow::Borrowed(self),
                service: Some(service),
            },
        };

        UpgradeableConnection {
            state: UpgradeableConnState::ReadVersion {
                read_version: read_version(io),
                builder: Cow::Borrowed(self),
                service: Some(service),
            },
        }

use std::convert::Infallible;
use std::time::Duration;

use bytes::Bytes;
use http_body_util::{BodyExt, Empty, Full};
use hyper::body::Incoming;
use hyper::client::conn::{http1 as client_h1, http2 as client_h2};
use hyper::service::service_fn;
use hyper::{Request, Response};
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto;
use tokio::net::{TcpListener, TcpStream};

async fn handler(req: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
  Ok(Response::new(Full::new(Bytes::from(format!("{:?}", req.version())))))
}

// One connection served by `builder`, with or without upgrades; returns the server's result.
async fn serve_one(
  builder: auto::Builder<TokioExecutor>,
  upgrades: bool,
  h2_client: bool,
) -> (String, String) {
  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 io = TokioIo::new(stream);
    let r = if upgrades {
      builder.serve_connection_with_upgrades(io, service_fn(handler)).await
    } else {
      builder.serve_connection(io, service_fn(handler)).await
    };
    match r {
      Ok(()) => "Ok(())".to_string(),
      Err(e) => format!("Err({})", chain(&*e)),
    }
  });
  let io = TokioIo::new(TcpStream::connect(addr).await.unwrap());
  let client = async {
    if h2_client {
      let (mut tx, conn) = client_h2::handshake::<_, _, Empty<Bytes>>(TokioExecutor::new(), io).await?;
      tokio::spawn(conn);
      let res = tx.send_request(Request::get("http://x/").body(Empty::new()).unwrap()).await?;
      res.into_body().collect().await.map(|c| c.to_bytes())
    } else {
      let (mut tx, conn) = client_h1::handshake::<_, Empty<Bytes>>(io).await?;
      tokio::spawn(conn);
      let res = tx.send_request(Request::get("/").body(Empty::new()).unwrap()).await?;
      res.into_body().collect().await.map(|c| c.to_bytes())
    }
  };
  let client = match tokio::time::timeout(Duration::from_secs(1), client).await {
    Ok(Ok(b)) => format!("{b:?}"),
    // the client-side error text races (canceled vs. broken pipe); keep only the fact
    Ok(Err(_)) => "error".to_string(),
    Err(_) => "timed out".to_string(),
  };
  let server = match tokio::time::timeout(Duration::from_secs(1), server).await {
    Ok(r) => r.unwrap(),
    Err(_) => "still running".to_string(),
  };
  (client, server)
}

fn chain(e: &(dyn std::error::Error + 'static)) -> String {
  let mut s = e.to_string();
  let mut src = e.source();
  while let Some(inner) = src {
    s.push_str(&format!(": {inner}"));
    src = inner.source();
  }
  s
}

#[tokio::main]
async fn main() {
  let builders: [(&str, fn() -> auto::Builder<TokioExecutor>); 2] = [
    ("http1_only", || auto::Builder::new(TokioExecutor::new()).http1_only()),
    ("http2_only", || auto::Builder::new(TokioExecutor::new()).http2_only()),
  ];
  for (name, make) in builders {
    for upgrades in [false, true] {
      let method = if upgrades { "serve_connection_with_upgrades" } else { "serve_connection" };
      for h2_client in [false, true] {
        let (client, server) = serve_one(make(), upgrades, h2_client).await;
        let who = if h2_client { "h2" } else { "h1" };
        println!("{name} + {method:<30} {who} client: {client:<11} server: {server}");
      }
    }
  }
}

http1_only + serve_connection               h1 client: b"HTTP/1.1" server: Ok(())
http1_only + serve_connection               h2 client: error       server: Err(invalid HTTP version parsed (found HTTP2 preface))
http1_only + serve_connection_with_upgrades h1 client: b"HTTP/1.1" server: Ok(())
http1_only + serve_connection_with_upgrades h2 client: b"HTTP/2.0" server: Ok(())
http2_only + serve_connection               h1 client: error       server: Err(http2 error: connection error detected: unspecific protocol error detected)
http2_only + serve_connection               h2 client: b"HTTP/2.0" server: Ok(())
http2_only + serve_connection_with_upgrades h1 client: b"HTTP/1.1" server: Ok(())
http2_only + serve_connection_with_upgrades h2 client: b"HTTP/2.0" server: Ok(())

pub fn downcast<T>(upgraded: Upgraded) -> Result<Parts<T>, Upgraded>
where
    T: Read + Write + Unpin + 'static,
{
    let hyper::upgrade::Parts {
        io: rewind,
        mut read_buf,
        ..
    } = upgraded.downcast::<Rewind<T>>()?;

    if let Some(pre) = rewind.pre {
        read_buf = if read_buf.is_empty() {
            pre
        } else {
            let mut buf = BytesMut::from(read_buf);

            buf.extend_from_slice(&pre);

            buf.freeze()
        };
    }

    Ok(Parts {
        io: rewind.inner,
        read_buf,
    })
}

use std::convert::Infallible;

use bytes::Bytes;
use http_body_util::Empty;
use hyper::body::Incoming;
use hyper::service::service_fn;
use hyper::{Request, Response, StatusCode};
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

async fn handler(mut req: Request<Incoming>) -> Result<Response<Empty<Bytes>>, Infallible> {
  let on = hyper::upgrade::on(&mut req);
  tokio::spawn(async move {
    let up = on.await.unwrap();
    // 1. hyper's own downcast, with the type we handed to serve_connection_with_upgrades
    let up = match up.downcast::<TokioIo<TcpStream>>() {
      Ok(_) => unreachable!(),
      Err(up) => {
        println!("hyper::upgrade::Upgraded::downcast::<TokioIo<TcpStream>>: Err");
        up
      }
    };
    // 2. hyper-util's downcast
    match auto::upgrade::downcast::<TokioIo<TcpStream>>(up) {
      Ok(parts) => {
        println!("auto::upgrade::downcast::<TokioIo<TcpStream>>: Ok, read_buf = {:?}", parts.read_buf);
        let mut tcp = parts.io.into_inner();
        tcp.write_all(b"bye").await.unwrap();
      }
      Err(_) => println!("auto::upgrade::downcast: Err"),
    }
  });
  Ok(
    Response::builder()
      .status(StatusCode::SWITCHING_PROTOCOLS)
      .header("connection", "upgrade")
      .header("upgrade", "foo")
      .body(Empty::new())
      .unwrap(),
  )
}

#[tokio::main]
async fn main() {
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  tokio::spawn(async move {
    let (stream, _) = listener.accept().await.unwrap();
    let builder = auto::Builder::new(TokioExecutor::new());
    builder
      .serve_connection_with_upgrades(TokioIo::new(stream), service_fn(handler))
      .await
      .unwrap();
  });
  let mut tcp = TcpStream::connect(addr).await.unwrap();
  // the request and the first bytes of the new protocol in one write
  tcp
    .write_all(b"GET / HTTP/1.1\r\nconnection: upgrade\r\nupgrade: foo\r\n\r\nhello")
    .await
    .unwrap();
  let mut out = String::new();
  tcp.read_to_string(&mut out).await.unwrap();
  println!("client read: {:?}", out.rsplit("\r\n").next().unwrap());
}

hyper::upgrade::Upgraded::downcast::<TokioIo<TcpStream>>: Err
auto::upgrade::downcast::<TokioIo<TcpStream>>: Ok, read_buf = b"hello"
client read: "bye"

http1::Builder:
  nothing                EOF at 503 ms          got ""  server: Err("read header from client timeout")
  "GET / HTTP/1.1\r\n"   EOF at 503 ms          got ""  server: Err("read header from client timeout")
  "PRI * HTTP/2.0\r\n"   EOF at 0 ms            got "HTTP/1.1 400 Bad Request"  server: Err("invalid HTTP version parsed")
  "PRI" then EOF         EOF at 0 ms            got ""  server: Err("connection closed before message completed")
auto::Builder:
  nothing                still open at 2000 ms  got ""  server: Ok(())
  "GET / HTTP/1.1\r\n"   EOF at 503 ms          got ""  server: Err("read header from client timeout")
  "PRI * HTTP/2.0\r\n"   still open at 2000 ms  got ""  server: Err("invalid HTTP version parsed")
  "PRI" then EOF         EOF at 0 ms            got ""  server: Err("connection closed before message completed")