Service 与 Body

pub trait Service<Request> {
    type Response;
    type Error;
    type Future: Future<Output = Result<Self::Response, Self::Error>>;

    /// Process the request and return the response asynchronously.
    /// `call` takes `&self` instead of `mut &self` because:
    /// - It prepares the way for async fn,
    ///   since then the future only borrows `&self`, and thus a Service can concurrently handle
    ///   multiple outstanding requests at once.
    /// - It's clearer that Services can likely be cloned.
    /// - To share state across clones, you generally need `Arc<Mutex<_>>`
    ///   That means you're not really using the `&mut self` and could do with a `&self`.
    ///   The discussion on this is here: <https://github.com/hyperium/hyper/issues/3040>
    fn call(&self, req: Request) -> Self::Future;
}

### Should there be a simplified `hyper::Service` trait, or should hyper depend on `tower-service`?

- There's still a few uncertain decisions around tower, such as if it should be
  changed to `async fn call`, and if `poll_ready` is the best way to handle
  backpressure.
- It's not clear that the backpressure is something needed at the `Server`
  boundary, thus meaning we should remove `poll_ready` from hyper.

        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), ()>> {
            if self.in_flight.is_some() {
                Poll::Pending
            } else {
                Poll::Ready(Ok(()))
            }
        }

pub trait HttpService<ReqBody>: sealed::Sealed<ReqBody> {
    type ResBody: Body;
    type Error: Into<Box<dyn StdError + Send + Sync>>;
    type Future: Future<Output = Result<Response<Self::ResBody>, Self::Error>>;

    #[doc(hidden)]
    fn call(&mut self, req: Request<ReqBody>) -> Self::Future;
}

impl<T, B1, B2> HttpService<B1> for T
where
    T: Service<Request<B1>, Response = Response<B2>>,
    B2: Body,
    T::Error: Into<Box<dyn StdError + Send + Sync>>,
{
    type ResBody = B2;

    type Error = T::Error;
    type Future = T::Future;

    fn call(&mut self, req: Request<B1>) -> Self::Future {
        Service::call(self, req)
    }
}

impl<F, ReqBody, Ret, ResBody, E> Service<Request<ReqBody>> for ServiceFn<F, ReqBody>
where
    F: Fn(Request<ReqBody>) -> Ret,
    ReqBody: Body,
    Ret: Future<Output = Result<Response<ResBody>, E>>,
    E: Into<Box<dyn StdError + Send + Sync>>,
    ResBody: Body,
{
    type Response = crate::Response<ResBody>;
    type Error = E;
    type Future = Ret;

    fn call(&self, req: Request<ReqBody>) -> Self::Future {
        (self.f)(req)
    }
}

impl<S, R> hyper::service::Service<R> for TowerToHyperService<S>
where
    S: tower_service::Service<R> + Clone,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = TowerToHyperServiceFuture<S, R>;

    fn call(&self, req: R) -> Self::Future {
        TowerToHyperServiceFuture {
            future: Oneshot::new(self.service.clone(), req),
        }
    }
}

            match this {
                OneshotProj::NotReady { svc, req } => {
                    ready!(svc.poll_ready(cx))?;
                    let fut = svc.call(req.take().expect("already called"));
                    self.set(Oneshot::Called { fut });
                }
                OneshotProj::Called { fut } => {
                    let res = ready!(fut.poll(cx))?;
                    self.set(Oneshot::Done);
                    return Poll::Ready(Ok(res));
                }
                OneshotProj::Done => panic!("polled after complete"),
            }

hyper = { version = "=1.12.0", features = ["server", "http1"] }
hyper-util = { version = "=0.1.21", features = ["tokio", "service"] }
tower = "=0.5.3"
http-body-util = "0.1"
bytes = "1"
tokio = { version = "=1.54.1", features = ["full"] }

use std::convert::Infallible;
use std::future::{ready, Ready};
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::task::{Context, Poll};

use bytes::Bytes;
use http_body_util::Full;
use hyper::{body::Incoming, server::conn::http1, Request, Response};
use hyper_util::{rt::TokioIo, service::TowerToHyperService};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

static CLONES: AtomicUsize = AtomicUsize::new(0);
static READY: AtomicUsize = AtomicUsize::new(0);
static CALLS: AtomicUsize = AtomicUsize::new(0);

// A tower service that counts clones, poll_ready and call.
struct Counted {
  hits: usize, // per-instance state
}

impl Clone for Counted {
  fn clone(&self) -> Self {
    CLONES.fetch_add(1, SeqCst);
    Counted { hits: self.hits }
  }
}

impl tower::Service<Request<Incoming>> for Counted {
  type Response = Response<Full<Bytes>>;
  type Error = Infallible;
  type Future = Ready<Result<Self::Response, Infallible>>;

  fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Infallible>> {
    READY.fetch_add(1, SeqCst);
    Poll::Ready(Ok(()))
  }

  fn call(&mut self, _req: Request<Incoming>) -> Self::Future {
    CALLS.fetch_add(1, SeqCst);
    self.hits += 1;
    ready(Ok(Response::new(Full::new(Bytes::from(format!("hits={}", self.hits))))))
  }
}

#[tokio::main]
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 = TowerToHyperService::new(Counted { hits: 0 });
    println!("after new:     clones={}", CLONES.load(SeqCst));
    http1::Builder::new().serve_connection(TokioIo::new(stream), svc).await.unwrap();
  });

  let mut client = TcpStream::connect(addr).await.unwrap();
  // three pipelined requests on one connection, the last one closes it
  client
    .write_all(
      b"GET / HTTP/1.1\r\nHost: x\r\n\r\n\
        GET / HTTP/1.1\r\nHost: x\r\n\r\n\
        GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n",
    )
    .await
    .unwrap();
  let mut buf = String::new();
  client.read_to_string(&mut buf).await.unwrap();
  server.await.unwrap();
  let bodies: Vec<&str> = buf.match_indices("hits=").map(|(i, _)| &buf[i..i + 6]).collect();
  println!("bodies:        {bodies:?}");
  println!(
    "after 3 reqs:  clones={} poll_ready={} call={}",
    CLONES.load(SeqCst),
    READY.load(SeqCst),
    CALLS.load(SeqCst)
  );
}

after new:     clones=0
bodies:        ["hits=1", "hits=1", "hits=1"]
after 3 reqs:  clones=3 poll_ready=3 call=3

pub trait Body {
    type Data: Buf;
    type Error;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>>;

    fn is_end_stream(&self) -> bool {
        false
    }

    fn size_hint(&self) -> SizeHint {
        SizeHint::default()
    }
}

                    let body_type = if body.is_end_stream() {
                        self.body_rx.set(None);
                        None
                    } else {
                        let btype = body
                            .size_hint()
                            .exact()
                            .map(BodyLength::Known)
                            .or(Some(BodyLength::Unknown));
                        self.body_rx.set(Some(body));
                        btype
                    };
                    self.conn.write_head(head, body_type);

HTTP/1.1 200 OK\r\n
content-length: 5\r\n
\r\n
helloHTTP/1.1 200 OK\r\n
content-length: 0\r\n
\r\n
HTTP/1.1 200 OK\r\n
content-length: 0\r\n
\r\n
HTTP/1.1 200 OK\r\n
connection: close\r\n
transfer-encoding: chunked\r\n
\r\n
5\r\n
hello\r\n
0\r\n
\r\n