uhttp/http1/
server.rs

1use std::convert::Infallible;
2use std::future::Future;
3use std::sync::Arc;
4
5use http_body_util::Full;
6use http_body_util::combinators::BoxBody;
7use hyper::Response as HyperResponse;
8use hyper::body::Bytes as HyperBytes;
9use hyper::service::service_fn;
10use hyper_util::rt::TokioExecutor;
11use hyper_util::rt::TokioIo;
12use hyper_util::server::conn::auto;
13use tokio::net::TcpListener;
14use tokio::net::ToSocketAddrs;
15
16use crate::Request;
17use crate::Response;
18
19pub struct Http1Server<F, Fut>
20where
21  F: 'static + Send + Sync + Fn(Request, Response) -> Fut,
22  Fut: 'static + Send + Future<Output = crate::Result<()>>,
23{
24  handle_func: Arc<F>,
25}
26
27impl<F, Fut> Http1Server<F, Fut>
28where
29  F: 'static + Send + Sync + Fn(Request, Response) -> Fut,
30  Fut: 'static + Send + Future<Output = crate::Result<()>>,
31{
32  pub fn new(handle_func: F) -> Self {
33    Self {
34      handle_func: Arc::new(handle_func),
35    }
36  }
37
38  pub async fn listen(
39    &self,
40    addr: impl ToSocketAddrs,
41  ) -> crate::Result<()> {
42    let listener: TcpListener = TcpListener::bind(&addr).await?;
43    let handler_func_ref = Arc::clone(&self.handle_func);
44
45    loop {
46      let Ok((stream, _)) = listener.accept().await else {
47        continue;
48      };
49
50      let handler_func_ref = handler_func_ref.clone();
51
52      tokio::task::spawn(async move {
53        let io = TokioIo::new(stream);
54
55        let service_builder = auto::Builder::new(TokioExecutor::new());
56        let service_handler = service_fn(move |req| {
57          let request = Request::new(req);
58
59          let (tx_res, rx_res) = tokio::sync::oneshot::channel();
60          let response = Response::new(tx_res, HyperResponse::builder());
61
62          let fut = handler_func_ref(request, response);
63          let (tx_fut_res, mut rx_fut_res) = tokio::sync::oneshot::channel::<crate::Result<()>>();
64
65          tokio::task::spawn(async move {
66            match fut.await {
67              Ok(_handler_response) => {}
68              Err(handler_error) => drop(tx_fut_res.send(Err(handler_error))),
69            };
70          });
71
72          async move {
73            Ok::<HyperResponse<BoxBody<HyperBytes, Infallible>>, crate::Error>(match rx_res.await {
74              Ok(res) => {
75                if let Ok(Err(err)) = rx_fut_res.try_recv() {
76                  return Ok(handle_error(crate::Error::generic(format!("{}", err))));
77                };
78                res
79              }
80              Err(err) => {
81                let x = handle_error(crate::Error::generic(format!("{}", err)));
82                x
83              }
84            })
85          }
86        });
87
88        #[cfg(feature = "websocket")]
89        service_builder
90          .serve_connection_with_upgrades(io, service_handler)
91          .await
92          .ok();
93
94        #[cfg(not(feature = "websocket"))]
95        service_builder
96          .serve_connection(io, service_handler)
97          .await
98          .ok();
99      });
100    }
101  }
102}
103
104fn handle_error(error: impl std::fmt::Display) -> HyperResponse<BoxBody<HyperBytes, Infallible>> {
105  let content = HyperBytes::from(format!("{}", error));
106  let body = BoxBody::new(Full::new(content));
107  let response = HyperResponse::builder().status(500).body(body);
108  let Ok(response) = response else { todo!() };
109  response
110}
uhttp/http1/server.rs

uhttp/http1/
server.rs
