use crate::lua::builtins::json::lua_value_to_json;
use http_body_util::Full;
use hyper::body::{Bytes, Frame, Incoming};
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{Request, Response, StatusCode};
use mlua::{Lua, Table, Value};
use std::cell::RefCell;
use std::collections::HashMap;
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};
use tokio::net::TcpListener;
use tracing::error;
#[derive(Clone)]
pub struct LuaAxumRouter(pub axum::Router);
impl mlua::UserData for LuaAxumRouter {}
pub(super) fn register_serve(lua: &Lua, http_table: &Table) -> mlua::Result<()> {
let serve_fn = lua.create_async_function(|lua, args: mlua::MultiValue| async move {
let mut args_iter = args.into_iter();
let port: u16 = match args_iter.next() {
Some(Value::Integer(n)) => n as u16,
_ => {
return Err::<(), _>(mlua::Error::runtime(
"http.serve: first argument must be a port number",
));
}
};
let routes_table = match args_iter.next() {
Some(Value::Table(t)) => t,
_ => {
return Err::<(), _>(mlua::Error::runtime(
"http.serve: second argument must be a routes table",
));
}
};
let routes = Rc::new(parse_routes(&routes_table)?);
let listener = TcpListener::bind(format!("0.0.0.0:{port}"))
.await
.map_err(|e| mlua::Error::runtime(format!("http.serve: bind failed: {e}")))?;
let actual_port = listener
.local_addr()
.map_err(|e| {
mlua::Error::runtime(format!("http.serve: failed to get local addr: {e}"))
})?
.port();
lua.globals().set("_SERVER_PORT", actual_port)?;
loop {
let (stream, addr) = listener
.accept()
.await
.map_err(|e| mlua::Error::runtime(format!("http.serve: accept failed: {e}")))?;
let peer_addr = addr.to_string();
let routes = routes.clone();
let lua_clone = lua.clone();
tokio::task::spawn_local(async move {
let io = hyper_util::rt::TokioIo::new(stream);
let routes = routes.clone();
let lua = lua_clone.clone();
let peer_addr = peer_addr.clone();
let service = service_fn(move |req: Request<Incoming>| {
let routes = routes.clone();
let lua = lua.clone();
let peer_addr = peer_addr.clone();
async move { handle_request(&lua, &routes, None, peer_addr, req).await }
});
if let Err(e) = http1::Builder::new()
.serve_connection(io, service)
.with_upgrades()
.await
&& !e.to_string().contains("connection closed")
{
error!("http.serve: connection error: {e}");
}
});
}
})?;
http_table.set("serve", serve_fn)?;
let serve_with_extra_fn =
lua.create_async_function(|lua, args: mlua::MultiValue| async move {
let mut args_iter = args.into_iter();
let port: u16 = match args_iter.next() {
Some(Value::Integer(n)) => n as u16,
_ => {
return Err::<(), _>(mlua::Error::runtime(
"http.serve_with_extra: first argument must be a port number",
));
}
};
let routes_table = match args_iter.next() {
Some(Value::Table(t)) => t,
_ => {
return Err::<(), _>(mlua::Error::runtime(
"http.serve_with_extra: second argument must be a routes table",
));
}
};
let extra_router: axum::Router = match args_iter.next() {
Some(Value::UserData(ud)) => {
let r = ud.borrow::<LuaAxumRouter>().map_err(|_| {
mlua::Error::runtime(
"http.serve_with_extra: third argument must be a LuaAxumRouter userdata",
)
})?;
r.0.clone()
}
_ => {
return Err::<(), _>(mlua::Error::runtime(
"http.serve_with_extra: third argument must be a LuaAxumRouter userdata",
));
}
};
let routes = Rc::new(parse_routes(&routes_table)?);
let listener = TcpListener::bind(format!("0.0.0.0:{port}"))
.await
.map_err(|e| {
mlua::Error::runtime(format!("http.serve_with_extra: bind failed: {e}"))
})?;
let actual_port = listener
.local_addr()
.map_err(|e| {
mlua::Error::runtime(format!(
"http.serve_with_extra: failed to get local addr: {e}"
))
})?
.port();
lua.globals().set("_SERVER_PORT", actual_port)?;
loop {
let (stream, addr) = listener.accept().await.map_err(|e| {
mlua::Error::runtime(format!("http.serve_with_extra: accept failed: {e}"))
})?;
let peer_addr = addr.to_string();
let routes = routes.clone();
let lua_clone = lua.clone();
let extra_router = extra_router.clone();
tokio::task::spawn_local(async move {
let io = hyper_util::rt::TokioIo::new(stream);
let routes = routes.clone();
let lua = lua_clone.clone();
let peer_addr = peer_addr.clone();
let extra_router = extra_router.clone();
let service = service_fn(move |req: Request<Incoming>| {
let routes = routes.clone();
let lua = lua.clone();
let peer_addr = peer_addr.clone();
let extra_router = extra_router.clone();
async move {
handle_request(&lua, &routes, Some(extra_router), peer_addr, req).await
}
});
if let Err(e) = http1::Builder::new()
.serve_connection(io, service)
.with_upgrades()
.await
&& !e.to_string().contains("connection closed")
{
error!("http.serve_with_extra: connection error: {e}");
}
});
}
})?;
http_table.set("serve_with_extra", serve_with_extra_fn)?;
Ok(())
}
struct SseBody {
rx: tokio::sync::mpsc::Receiver<Bytes>,
}
impl hyper::body::Body for SseBody {
type Data = Bytes;
type Error = std::convert::Infallible;
fn poll_frame(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
match self.rx.poll_recv(cx) {
Poll::Ready(Some(bytes)) => Poll::Ready(Some(Ok(Frame::data(bytes)))),
Poll::Ready(None) => Poll::Ready(None),
Poll::Pending => Poll::Pending,
}
}
}
fn format_sse_event(event_table: &Table) -> mlua::Result<String> {
let mut out = String::new();
if let Ok(Some(event)) = event_table.get::<Option<String>>("event") {
if event.contains('\n') || event.contains('\r') {
return Err(mlua::Error::runtime(
"SSE event name must not contain newlines",
));
}
out.push_str("event: ");
out.push_str(&event);
out.push('\n');
}
if let Ok(Some(data)) = event_table.get::<Option<String>>("data") {
for line in data.split('\n') {
out.push_str("data: ");
out.push_str(line);
out.push('\n');
}
}
if let Ok(Some(id)) = event_table.get::<Option<String>>("id") {
if id.contains('\n') || id.contains('\r') {
return Err(mlua::Error::runtime("SSE id must not contain newlines"));
}
out.push_str("id: ");
out.push_str(&id);
out.push('\n');
}
if let Ok(Some(retry)) = event_table.get::<Option<i64>>("retry") {
out.push_str("retry: ");
out.push_str(&retry.to_string());
out.push('\n');
}
out.push('\n');
Ok(out)
}
fn parse_routes(routes_table: &Table) -> mlua::Result<HashMap<(String, String), mlua::Function>> {
let mut routes = HashMap::new();
for method_pair in routes_table.pairs::<String, Table>() {
let (method, paths_table) = method_pair?;
let method_upper = method.to_uppercase();
for path_pair in paths_table.pairs::<String, mlua::Function>() {
let (path, func) = path_pair?;
routes.insert((method_upper.clone(), path), func);
}
}
Ok(routes)
}
type ServerBody = axum::body::Body;
fn lookup_route<'a>(
routes: &'a HashMap<(String, String), mlua::Function>,
method: &str,
path: &str,
) -> Option<&'a mlua::Function> {
let key = (method.to_string(), path.to_string());
if let Some(f) = routes.get(&key) {
return Some(f);
}
let mut search = path;
while let Some(pos) = search.rfind('/') {
let prefix = &search[..pos];
let wildcard_key = (method.to_string(), format!("{prefix}/*"));
if let Some(f) = routes.get(&wildcard_key) {
return Some(f);
}
if pos == 0 {
let root_key = (method.to_string(), "/*".to_string());
return routes.get(&root_key);
}
search = prefix;
}
None
}
fn is_websocket_upgrade(headers: &[(String, String)]) -> bool {
headers.iter().any(|(k, v)| {
k.eq_ignore_ascii_case("upgrade") && v.to_ascii_lowercase().contains("websocket")
})
}
fn validate_ws_request(headers: &[(String, String)]) -> Result<String, &'static str> {
let mut has_connection_upgrade = false;
let mut version_ok = false;
let mut key: Option<String> = None;
for (k, v) in headers {
match k.to_ascii_lowercase().as_str() {
"connection" if v.to_ascii_lowercase().contains("upgrade") => {
has_connection_upgrade = true;
}
"sec-websocket-version" if v.trim() == "13" => {
version_ok = true;
}
"sec-websocket-key" => {
key = Some(v.clone());
}
_ => {}
}
}
if !has_connection_upgrade {
return Err("missing Connection: Upgrade header");
}
if !version_ok {
return Err("Sec-WebSocket-Version must be 13");
}
key.ok_or("missing Sec-WebSocket-Key header")
}
fn compute_ws_accept(key: &str) -> String {
use sha1::Digest;
const MAGIC: &[u8] = b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
let mut hasher = sha1::Sha1::new();
hasher.update(key.as_bytes());
hasher.update(MAGIC);
let digest = hasher.finalize();
data_encoding::BASE64.encode(&digest)
}
async fn forward_to_axum_router(
mut router: axum::Router,
req: Request<Incoming>,
) -> Result<Response<ServerBody>, hyper::Error> {
use tower::Service;
match <axum::Router as Service<Request<Incoming>>>::call(&mut router, req).await {
Ok(resp) => Ok(resp),
Err(_) => unreachable!("axum::Router::call is Infallible"),
}
}
async fn handle_request(
lua: &Lua,
routes: &HashMap<(String, String), mlua::Function>,
extra_router: Option<axum::Router>,
peer_addr: String,
req: Request<Incoming>,
) -> Result<Response<ServerBody>, hyper::Error> {
let method = req.method().to_string();
let path = req.uri().path().to_string();
let query = req.uri().query().unwrap_or("").to_string();
let headers: Vec<(String, String)> = req
.headers()
.iter()
.filter_map(|(k, v)| v.to_str().ok().map(|v| (k.to_string(), v.to_string())))
.collect();
let is_ws = is_websocket_upgrade(&headers);
let handler = match lookup_route(routes, &method, &path) {
Some(h) => h.clone(),
None => {
if let Some(router) = extra_router {
return forward_to_axum_router(router, req).await;
}
return Ok(Response::builder()
.status(StatusCode::NOT_FOUND)
.header("content-type", "text/plain")
.body(axum::body::Body::new(Full::new(Bytes::from("not found"))))
.unwrap());
}
};
if is_ws {
let lua_resp =
match build_lua_request_and_call(lua, &handler, &method, &path, &query, &headers, "")
.await
{
Ok(t) => t,
Err(e) => {
return Ok(Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.header("content-type", "text/plain")
.body(axum::body::Body::new(Full::new(Bytes::from(format!(
"handler error: {e}"
)))))
.unwrap());
}
};
if let Ok(Some(ws_fn)) = lua_resp.get::<Option<mlua::Function>>("ws") {
return build_ws_upgrade_response(lua, &headers, lua_resp, ws_fn, peer_addr, req);
}
return lua_response_to_http(lua, &lua_resp);
}
let body_bytes = match http_body_util::BodyExt::collect(req.into_body()).await {
Ok(collected) => collected.to_bytes(),
Err(_) => Bytes::new(),
};
let body_str = String::from_utf8_lossy(&body_bytes).to_string();
match build_lua_request_and_call(lua, &handler, &method, &path, &query, &headers, &body_str)
.await
{
Ok(lua_resp) => lua_response_to_http(lua, &lua_resp),
Err(e) => Ok(Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.header("content-type", "text/plain")
.body(axum::body::Body::new(Full::new(Bytes::from(format!(
"handler error: {e}"
)))))
.unwrap()),
}
}
fn build_ws_upgrade_response(
lua: &Lua,
headers: &[(String, String)],
resp_table: Table,
ws_fn: mlua::Function,
peer_addr: String,
req: Request<Incoming>,
) -> Result<Response<ServerBody>, hyper::Error> {
let key = match validate_ws_request(headers) {
Ok(k) => k,
Err(msg) => {
return Ok(Response::builder()
.status(StatusCode::BAD_REQUEST)
.header("content-type", "text/plain")
.body(axum::body::Body::new(Full::new(Bytes::from(format!(
"websocket upgrade rejected: {msg}"
)))))
.unwrap());
}
};
let accept = compute_ws_accept(&key);
let mut builder = Response::builder()
.status(StatusCode::SWITCHING_PROTOCOLS)
.header(hyper::header::UPGRADE, "websocket")
.header(hyper::header::CONNECTION, "Upgrade")
.header("sec-websocket-accept", accept);
if let Ok(Some(headers_table)) = resp_table.get::<Option<Table>>("headers") {
for pair in headers_table.pairs::<String, mlua::String>().flatten() {
let (k, v) = pair;
let kl = k.to_ascii_lowercase();
if matches!(
kl.as_str(),
"upgrade" | "connection" | "sec-websocket-accept"
) {
continue;
}
if let Ok(s) = v.to_str() {
builder = builder.header(&k, s.as_ref());
}
}
}
let response = builder
.body(axum::body::Body::new(Full::new(Bytes::new())))
.unwrap();
let lua_clone = lua.clone();
tokio::task::spawn_local(async move {
let upgraded = match hyper::upgrade::on(req).await {
Ok(u) => u,
Err(e) => {
error!("http.serve: ws upgrade failed: {e}");
return;
}
};
let io = hyper_util::rt::TokioIo::new(upgraded);
let stream = tokio_tungstenite::WebSocketStream::from_raw_socket(
io,
tokio_tungstenite::tungstenite::protocol::Role::Server,
None,
)
.await;
let conn = crate::lua::builtins::ws::WsServerConn::new(stream, peer_addr);
let ud = match lua_clone.create_userdata(conn) {
Ok(u) => u,
Err(e) => {
error!("http.serve: ws userdata creation failed: {e}");
return;
}
};
if let Err(e) = ws_fn.call_async::<()>(ud).await
&& !e.to_string().contains("conn:read: ")
{
error!("http.serve: ws handler error: {e}");
}
});
Ok(response)
}
async fn build_lua_request_and_call(
lua: &Lua,
handler: &mlua::Function,
method: &str,
path: &str,
query: &str,
headers: &[(String, String)],
body: &str,
) -> mlua::Result<Table> {
let req_table = lua.create_table()?;
req_table.set("method", method.to_string())?;
req_table.set("path", path.to_string())?;
req_table.set("query", query.to_string())?;
req_table.set("body", body.to_string())?;
let params_table = lua.create_table()?;
if !query.is_empty() {
for (key, value) in url::form_urlencoded::parse(query.as_bytes()) {
params_table.set(key.into_owned(), value.into_owned())?;
}
}
req_table.set("params", params_table)?;
let headers_table = lua.create_table()?;
for (k, v) in headers {
headers_table.set(k.as_str(), v.as_str())?;
}
req_table.set("headers", headers_table)?;
handler.call_async::<Table>(req_table).await
}
fn lua_response_to_http(
lua: &Lua,
resp_table: &Table,
) -> Result<Response<ServerBody>, hyper::Error> {
let status = resp_table
.get::<Option<u16>>("status")
.unwrap_or(None)
.unwrap_or(200);
if let Ok(Some(sse_fn)) = resp_table.get::<Option<mlua::Function>>("sse") {
let (tx, rx) = tokio::sync::mpsc::channel::<Bytes>(32);
let mut builder =
Response::builder().status(StatusCode::from_u16(status).unwrap_or(StatusCode::OK));
if let Ok(Some(headers_table)) = resp_table.get::<Option<Table>>("headers") {
for pair in headers_table.pairs::<String, Value>().flatten() {
let (k, v) = pair;
match v {
Value::String(s) => {
if let Ok(s) = s.to_str() {
builder = builder.header(&k, s.as_ref());
}
}
Value::Table(t) => {
for val in t.sequence_values::<String>().flatten() {
builder = builder.header(&k, val);
}
}
_ => {}
}
}
}
let mut response = builder.body(axum::body::Body::new(SseBody { rx })).unwrap();
let response_headers = response.headers_mut();
if !response_headers.contains_key(hyper::header::CONTENT_TYPE) {
response_headers.insert(
hyper::header::CONTENT_TYPE,
hyper::header::HeaderValue::from_static("text/event-stream"),
);
}
if !response_headers.contains_key(hyper::header::CACHE_CONTROL) {
response_headers.insert(
hyper::header::CACHE_CONTROL,
hyper::header::HeaderValue::from_static("no-cache"),
);
}
if !response_headers.contains_key(hyper::header::CONNECTION) {
response_headers.insert(
hyper::header::CONNECTION,
hyper::header::HeaderValue::from_static("keep-alive"),
);
}
let lua_clone = lua.clone();
tokio::task::spawn_local(async move {
let tx_holder: Rc<RefCell<Option<tokio::sync::mpsc::Sender<Bytes>>>> =
Rc::new(RefCell::new(Some(tx)));
let tx_for_fn = tx_holder.clone();
let send_fn = match lua_clone.create_async_function(move |_lua, event_table: Table| {
let tx_ref = tx_for_fn.clone();
async move {
let formatted = format_sse_event(&event_table)?;
let tx = tx_ref
.borrow()
.clone()
.ok_or_else(|| mlua::Error::runtime("SSE stream closed"))?;
if tx.send(Bytes::from(formatted)).await.is_err() {
return Err(mlua::Error::runtime("SSE stream closed"));
}
Ok(())
}
}) {
Ok(f) => f,
Err(e) => {
error!("http.serve SSE: failed to create send callback: {e}");
return;
}
};
if let Err(e) = sse_fn.call_async::<()>(send_fn).await
&& !e.to_string().contains("SSE stream closed")
{
error!("http.serve SSE: handler error: {e}");
}
tx_holder.borrow_mut().take();
});
return Ok(response);
}
let mut builder =
Response::builder().status(StatusCode::from_u16(status).unwrap_or(StatusCode::OK));
let has_content_type =
if let Ok(Some(headers_table)) = resp_table.get::<Option<Table>>("headers") {
let mut found_ct = false;
for pair in headers_table.pairs::<String, Value>().flatten() {
let (k, v) = pair;
if k.eq_ignore_ascii_case("content-type") {
found_ct = true;
}
match v {
Value::String(s) => {
if let Ok(s) = s.to_str() {
builder = builder.header(&k, s.as_ref());
}
}
Value::Table(t) => {
for val in t.sequence_values::<String>().flatten() {
builder = builder.header(&k, val);
}
}
_ => {}
}
}
found_ct
} else {
false
};
let body_bytes = if let Ok(Some(json_table)) = resp_table.get::<Option<Table>>("json") {
let json_val =
lua_value_to_json(&Value::Table(json_table)).unwrap_or(serde_json::Value::Null);
let serialized = serde_json::to_string(&json_val).unwrap_or_else(|_| "null".to_string());
if !has_content_type {
builder = builder.header("content-type", "application/json");
}
Bytes::from(serialized)
} else if let Ok(Some(body_lua)) = resp_table.get::<Option<mlua::String>>("body") {
if !has_content_type {
builder = builder.header("content-type", "text/plain");
}
Bytes::from(body_lua.as_bytes().to_vec())
} else {
if !has_content_type {
builder = builder.header("content-type", "text/plain");
}
Bytes::new()
};
Ok(builder
.body(axum::body::Body::new(Full::new(body_bytes)))
.unwrap())
}
#[cfg(all(test, feature = "server"))]
mod tests {
use super::*;
use axum::Router;
use axum::routing::get;
use mlua::Lua;
#[test]
fn lua_axum_router_round_trips_through_mlua_globals() {
let lua = Lua::new();
let router = Router::new().route("/ping", get(|| async { "pong" }));
let wrapped = LuaAxumRouter(router);
let ud = lua
.create_userdata(wrapped)
.expect("create_userdata for LuaAxumRouter");
lua.globals()
.set("EXTRA_ROUTER", ud)
.expect("stash userdata in globals");
let value: mlua::Value = lua
.globals()
.get("EXTRA_ROUTER")
.expect("read userdata back from globals");
let ud = match value {
mlua::Value::UserData(u) => u,
other => panic!("expected UserData, got {other:?}"),
};
let _borrowed = ud
.borrow::<LuaAxumRouter>()
.expect("downcast to LuaAxumRouter");
}
#[test]
fn lua_axum_router_is_clone_and_preserves_routes() {
let router = Router::<()>::new().route("/health", get(|| async { "ok" }));
let wrapped = LuaAxumRouter(router);
let _cloned = wrapped.clone();
}
}