use std::net::SocketAddr;
use tokio::net::TcpListener;
use hyper::server::conn::http1;
use hyper_util::rt::TokioIo;
use hyper_util::service::TowerToHyperService;
use crate::error::{Error, Result};
use crate::types::{OxiditeRequest, OxiditeResponse};
use tower_service::Service;
use std::error::Error as StdError;
use http_body_util::BodyExt;
use std::task::{Context, Poll};
#[cfg(feature = "http3")]
pub mod http3_server;
#[cfg(feature = "http3")]
pub use http3_server::Http3Server;
#[derive(Clone)]
pub struct BodyAdapter<S>(S);
impl<S> BodyAdapter<S> {
pub fn new(service: S) -> Self {
Self(service)
}
}
use std::pin::Pin;
impl<S> Service<hyper::Request<hyper::body::Incoming>> for BodyAdapter<S>
where
S: Service<OxiditeRequest, Response = OxiditeResponse, Error = Error> + Clone + Send + 'static,
S::Future: Send + 'static,
{
type Response = hyper::Response<crate::types::BoxBody>;
type Error = Error;
type Future = Pin<Box<dyn std::future::Future<Output = std::result::Result<Self::Response, Self::Error>> + Send>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
self.0.poll_ready(cx)
}
fn call(&mut self, req: hyper::Request<hyper::body::Incoming>) -> Self::Future {
let accepts_html = req.headers().get(hyper::header::ACCEPT)
.map(|h| h.to_str().unwrap_or("").contains("text/html"))
.unwrap_or(false);
let req = req.map(|b| b.map_err(|e| e.into()).boxed());
let fut = self.0.call(req);
Box::pin(async move {
match fut.await {
Ok(response) => Ok(response.into()),
Err(error) => {
let env = std::env::var("OXIDITE_ENV").unwrap_or_else(|_| "development".to_string());
if env == "development" && accepts_html && error.is_server_error() {
use bytes::Bytes;
use http_body_util::Full;
use hyper::header::{CONTENT_TYPE, SERVER};
let html = crate::error::render_ignition_error(&error);
let res = hyper::Response::builder()
.status(error.status_code())
.header(CONTENT_TYPE, "text/html; charset=utf-8")
.header(SERVER, crate::response::SERVER_HEADER_VALUE)
.body(Full::new(Bytes::from(html)).map_err(|e| match e {}).boxed())
.unwrap();
Ok(res)
} else {
Ok(OxiditeResponse::from(error).into())
}
}
}
})
}
}
pub struct Server<S> {
service: S,
}
impl<S> Server<S>
where
S: Service<OxiditeRequest, Response = OxiditeResponse, Error = Error> + Clone + Send + Sync + 'static,
S::Future: Send + 'static,
{
pub fn new(service: S) -> Self {
Self {
service,
}
}
pub async fn listen(self, addr: SocketAddr) -> Result<()> {
let listener = TcpListener::bind(addr).await?;
println!("Listening on http://{}", addr);
loop {
let (stream, _) = listener.accept().await?;
let io = TokioIo::new(stream);
let service = self.service.clone();
tokio::task::spawn(async move {
let service = BodyAdapter::new(service);
let hyper_service = TowerToHyperService::new(service);
if let Err(err) = http1::Builder::new()
.serve_connection(io, hyper_service)
.await
{
if let Some(service_err) = err.source().and_then(|e| e.downcast_ref::<Error>()) {
if service_err.is_server_error() {
eprintln!("Server error: {}", service_err);
}
} else {
let err_msg = err.to_string();
if !err_msg.contains("NotFound") && !err_msg.contains("connection closed") {
eprintln!("Connection error: {}", err);
}
}
}
});
}
}
#[cfg(feature = "http3")]
pub async fn listen_h3(self, addr: SocketAddr, cert_pem: &str, key_pem: &str) -> Result<()> {
use rustls::ServerConfig;
use rustls_pemfile::{certs, pkcs8_private_keys};
use std::io::Cursor;
let http1_addr = addr;
let http1_service = self.service.clone();
tokio::spawn(async move {
let listener = TcpListener::bind(http1_addr).await.unwrap();
println!("HTTP/1.1 server listening on http://{}", http1_addr);
loop {
let (stream, _) = listener.accept().await.unwrap();
let io = TokioIo::new(stream);
let service = http1_service.clone();
tokio::task::spawn(async move {
let service = BodyAdapter::new(service);
let hyper_service = TowerToHyperService::new(service);
if let Err(err) = http1::Builder::new()
.serve_connection(io, hyper_service)
.await
{
if let Some(service_err) = err.source().and_then(|e| e.downcast_ref::<Error>()) {
if service_err.is_server_error() {
eprintln!("HTTP/1.1 server error: {}", service_err);
}
} else {
let err_msg = err.to_string();
if !err_msg.contains("NotFound") && !err_msg.contains("connection closed") {
eprintln!("HTTP/1.1 connection error: {}", err);
}
}
}
});
}
});
let cert_chain = certs(&mut Cursor::new(cert_pem))
.collect::<std::result::Result<Vec<_>, _>>()
.map_err(|e| crate::error::Error::InternalServerError(e.to_string()))?;
let mut keys = pkcs8_private_keys(&mut Cursor::new(key_pem))
.collect::<std::result::Result<Vec<_>, _>>()?;
if keys.is_empty() {
return Err(crate::error::Error::InternalServerError("No private keys found".to_string()));
}
let tls_config = ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(cert_chain, rustls::pki_types::PrivateKeyDer::Pkcs8(keys.remove(0)))
.map_err(|e| crate::error::Error::InternalServerError(e.to_string()))?;
let http3_server = Http3Server::new(self.service);
http3_server.listen(addr, tls_config).await?;
Ok(())
}
}