A pit full of rusty nails
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586use std::rc::Rc;
use futures_concurrency::future::Race;use hyper_util::{ rt::TokioIo, server::{conn::auto::Builder, graceful::GracefulShutdown},};use nailbox::boxed_future_within;use nailcancel::CancelToken;use nailip::IdentifyPeer;use nailrater::{LimitModes, NailRater};use nailresponder::NailResponder;use nailrt::HyperExecutor;use nailservice::normalize::NormalizeUrls;use nailstate::ServerState;use nailtrace::TraceConnection;use tokio::net::TcpListener;
async fn server_loop( listener: TcpListener, app: ServerState, graceful: &GracefulShutdown,) -> color_eyre::Result<()> { let server = Builder::new(HyperExecutor::new());
let is_traced = app.config.open_telemetry.logs || app.config.open_telemetry.traces || app.config.open_telemetry.record_details;
let partial_service = Rc::new(NormalizeUrls::new(Rc::new(TraceConnection::new( is_traced, Rc::new(NailRater::new( LimitModes::from(&app.config.rate_limiting), app.spicy_payloads.get(), app.config.clone_inner(), Rc::new(NailResponder::new(app)), )), ))));
loop { let (socket, remote_addr) = listener.accept().await?;
let socket = TokioIo::new(socket);
let hyper_service = IdentifyPeer::new(remote_addr, partial_service.clone());
let conn = boxed_future_within(|| { let conn = server.serve_connection(socket, hyper_service);
graceful.watch(conn.into_owned()) });
tokio::task::spawn_local(conn); }}
async fn cancel_loop(token: CancelToken) -> color_eyre::Result<()> { token.cancelled().await; Ok(())}
/// Serves a `hyper` service app, taking a [`CancelToken`] to do a graceful shutdown/// loop.pub async fn serve(app: ServerState, token: CancelToken) -> color_eyre::Result<()> { let listener = nailnet::get_tcp_socket(&app.config.server.socket_addr)?; let ip = listener.local_addr()?;
tracing::info!( port = ip.port(), address = ip.ip().to_string(), "{} listening on {}", std::thread::current().name().unwrap(), ip );
let graceful = GracefulShutdown::new();
(server_loop(listener, app, &graceful), cancel_loop(token)) .race() .await?;
graceful.shutdown().await;
Ok(())}