ec0bdb8
use std::convert::Infallible;
use std::io;
use std::time::Duration;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper_util::rt::{TokioIo, TokioTimer};
use tokio::net::TcpListener;
use tokio::sync::watch;
use tokio::task::JoinSet;
use super::SocketHub;
use super::handle::handle;
impl SocketHub {
pub async fn serve(
&self,
listener: TcpListener,
mut shutdown: watch::Receiver<bool>,
) -> io::Result<()> {
let mut connections = JoinSet::new();
let mut stopped = self.shutdown_receiver();
let result = loop {
if *shutdown.borrow() || *stopped.borrow() {
break Ok(());
}
tokio::select! {
_ = shutdown.changed() => break Ok(()),
_ = stopped.changed() => break Ok(()),
accepted = listener.accept() => {
let (stream, _) = match accepted {
Ok(connection) => connection,
Err(error) => break Err(error),
};
let hub = self.clone();
let mut stop = self.shutdown_receiver();
connections.spawn(async move {
let service = service_fn(move |request| {
let hub = hub.clone();
async move { Ok::<_, Infallible>(handle(request, hub).await) }
});
let connection = http1::Builder::new()
.timer(TokioTimer::new())
.header_read_timeout(Duration::from_secs(10))
.max_buf_size(32 * 1024)
.serve_connection(TokioIo::new(stream), service)
.with_upgrades();
tokio::pin!(connection);
if *stop.borrow() {
return;
}
tokio::select! {
result = &mut connection => {
if let Err(error) = result {
eprintln!("kenevent: HTTP connection failed: {error}");
}
}
_ = stop.changed() => {
connection.as_mut().graceful_shutdown();
let _ = tokio::time::timeout(Duration::from_secs(5), &mut connection).await;
}
}
});
}
Some(result) = connections.join_next(), if !connections.is_empty() => {
if let Err(error) = result {
eprintln!("kenevent: HTTP task failed: {error}");
}
}
}
};
drop(listener);
self.shutdown();
if tokio::time::timeout(Duration::from_secs(5), async {
while connections.join_next().await.is_some() {}
})
.await
.is_err()
{
connections.abort_all();
while connections.join_next().await.is_some() {}
}
self.finish_transports().await;
result
}
}