diff options
author | 19年梦醒 <3949379+getong@users.noreply.github.com> | 2024-03-22 07:06:00 +0800 |
---|---|---|
committer | GitHub <noreply@github.com> | 2024-03-21 23:06:00 +0000 |
commit | 849206ef9d56b250424f34077a62a4e6ad96c21a (patch) | |
tree | 4185c2418e8173bb8dbc11e2d315851c22f0af63 | |
parent | 80fff4f2e7f8250c28bc6f5de79dace1ec914b31 (diff) | |
download | mlua-849206ef9d56b250424f34077a62a4e6ad96c21a.zip |
update hyper to v1, and add shell command example (#384)
Co-authored-by: Alex Orlenko <zxteam@protonmail.com>
-rw-r--r-- | Cargo.toml | 5 | ||||
-rw-r--r-- | README.md | 11 | ||||
-rw-r--r-- | examples/async_http_client.rs | 53 | ||||
-rw-r--r-- | examples/async_http_server.rs | 111 |
4 files changed, 106 insertions, 74 deletions
@@ -63,7 +63,10 @@ libloading = { version = "0.8", optional = true } [dev-dependencies] trybuild = "1.0" futures = "0.3.5" -hyper = { version = "0.14", features = ["client", "server"] } +hyper = { version = "1", features = ["client", "server"] } +hyper-util = { version = "0.1.3", features = ["server", "client", "client-legacy", "tokio", "http1"] } +http-body-util = "0.1.1" +bytes = "1.5.0" reqwest = { version = "0.12", features = ["json"] } tokio = { version = "1.0", features = ["macros", "rt", "time"] } serde = { version = "1.0", features = ["derive"] } @@ -84,6 +84,17 @@ This works using Lua [coroutines](https://www.lua.org/manual/5.3/manual.html#2.6 - [HTTP Server](examples/async_http_server.rs) - [TCP Server](examples/async_tcp_server.rs) + +**shell command example**: +``` shell +# async http client +cargo run --example async_http_client --features="async macros lua54" + +# async http server +cargo run --example async_http_server --features="async macros lua54" +curl http://localhost:3000 +``` + ### Serialization (serde) support With `serialize` feature flag enabled, `mlua` allows you to serialize/deserialize any type that implements [`serde::Serialize`] and [`serde::Deserialize`] into/from [`mlua::Value`]. In addition `mlua` provides [`serde::Serialize`] trait implementation for it (including `UserData` support). diff --git a/examples/async_http_client.rs b/examples/async_http_client.rs index 3eccdc3..c0fa1ff 100644 --- a/examples/async_http_client.rs +++ b/examples/async_http_client.rs @@ -1,20 +1,38 @@ +use bytes::Bytes; +use http_body_util::BodyExt; +use http_body_util::Empty; +use hyper::body::Incoming; +use hyper_util::client::legacy::Client as HyperClient; use std::collections::HashMap; -use hyper::body::{Body as HyperBody, HttpBody as _}; -use hyper::Client as HyperClient; - use mlua::{chunk, ExternalResult, Lua, Result, UserData, UserDataMethods}; -struct BodyReader(HyperBody); +struct BodyReader(Incoming); impl UserData for BodyReader { fn add_methods<'lua, M: UserDataMethods<'lua, Self>>(methods: &mut M) { methods.add_async_method_mut("read", |lua, reader, ()| async move { - if let Some(bytes) = reader.0.data().await { - let bytes = bytes.into_lua_err()?; - return Some(lua.create_string(&bytes)).transpose(); + let mut summarize = Vec::new(); // Create a vector to accumulate the bytes + + loop { + match reader.0.frame().await { + Some(Ok(bytes)) => { + if let Ok(data) = bytes.into_data() { + summarize.extend(data); // Append the bytes to the summarize variable + } + } + Some(Err(_)) => break, // Break on error + None => break, // Break if no more frames + } + } + + if !summarize.is_empty() { + // If summarize has collected data, return it as a Lua string + Ok(Some(lua.create_string(&summarize)?)) + } else { + // Return None if no data was collected + Ok(None) } - Ok(None) }); } } @@ -24,7 +42,8 @@ async fn main() -> Result<()> { let lua = Lua::new(); let fetch_url = lua.create_async_function(|lua, uri: String| async move { - let client = HyperClient::new(); + let client = + HyperClient::builder(hyper_util::rt::TokioExecutor::new()).build_http::<Empty<Bytes>>(); let uri = uri.parse().into_lua_err()?; let resp = client.get(uri).await.into_lua_err()?; @@ -47,18 +66,18 @@ async fn main() -> Result<()> { let f = lua .load(chunk! { - local res = $fetch_url(...) + local res = $fetch_url(...) print("status: "..res.status) for key, vals in pairs(res.headers) do - for _, val in ipairs(vals) do - print(key..": "..val) - end + for _, val in ipairs(vals) do + print(key..": "..val) + end end repeat - local body = res.body:read() - if body then - print(body) - end + local body = res.body:read() + if body then + print(body) + end until not body }) .into_function()?; diff --git a/examples/async_http_server.rs b/examples/async_http_server.rs index 43ae7a9..cd94e6e 100644 --- a/examples/async_http_server.rs +++ b/examples/async_http_server.rs @@ -1,18 +1,14 @@ -use std::future::Future; -use std::net::SocketAddr; -use std::pin::Pin; -use std::rc::Rc; -use std::task::{Context, Poll}; - -use hyper::server::conn::AddrStream; -use hyper::service::Service; -use hyper::{Body, Request, Response, Server}; - +use bytes::Bytes; +use http_body_util::{combinators::BoxBody, BodyExt, Empty, Full}; +use hyper::{body::Incoming, service::Service, Request, Response}; +use hyper_util::{rt::TokioIo, server::conn::auto}; use mlua::{ chunk, Error as LuaError, Function, Lua, String as LuaString, Table, UserData, UserDataMethods, }; +use std::{future::Future, net::SocketAddr, pin::Pin, rc::Rc}; +use tokio::{net::TcpListener, task::LocalSet}; -struct LuaRequest(SocketAddr, Request<Body>); +struct LuaRequest(SocketAddr, Request<Incoming>); impl UserData for LuaRequest { fn add_methods<'lua, M: UserDataMethods<'lua, Self>>(methods: &mut M) { @@ -23,16 +19,12 @@ impl UserData for LuaRequest { pub struct Svc(Rc<Lua>, SocketAddr); -impl Service<Request<Body>> for Svc { - type Response = Response<Body>; +impl Service<Request<Incoming>> for Svc { + type Response = Response<BoxBody<Bytes, hyper::Error>>; type Error = LuaError; type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>; - fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { - Poll::Ready(Ok(())) - } - - fn call(&mut self, req: Request<Body>) -> Self::Future { + fn call(&self, req: Request<Incoming>) -> Self::Future { // If handler returns an error then generate 5xx response let lua = self.0.clone(); let lua_req = LuaRequest(self.1, req); @@ -53,8 +45,16 @@ impl Service<Request<Body>> for Svc { let body = lua_resp .get::<_, Option<LuaString>>("body")? - .map(|b| Body::from(b.as_bytes().to_vec())) - .unwrap_or_else(Body::empty); + .map(|b| { + Full::new(Bytes::copy_from_slice(b.clone().as_bytes())) + .map_err(|never| match never {}) + .boxed() + }) + .unwrap_or_else(|| { + Empty::<Bytes>::new() + .map_err(|never| match never {}) + .boxed() + }); Ok(resp.body(body).unwrap()) } @@ -62,7 +62,11 @@ impl Service<Request<Body>> for Svc { eprintln!("{}", err); Ok(Response::builder() .status(500) - .body(Body::from("Internal Server Error")) + .body( + Full::new(Bytes::from("Internal Server Error".as_bytes())) + .map_err(|never| match never {}) + .boxed(), + ) .unwrap()) } } @@ -77,16 +81,16 @@ async fn main() { // Create Lua handler function let handler: Function = lua .load(chunk! { - function(req) - return { - status = 200, - headers = { - ["X-Req-Method"] = req:method(), - ["X-Remote-Addr"] = req:remote_addr(), - }, - body = "Hello from Lua!\n" - } - end + function(req) + return { + status = 200, + headers = { + ["X-Req-Method"] = req:method(), + ["X-Remote-Addr"] = req:remote_addr(), + }, + body = "Hello from Lua!\n" + } + end }) .eval() .expect("cannot create Lua handler"); @@ -95,31 +99,26 @@ async fn main() { lua.set_named_registry_value("http_handler", handler) .expect("cannot store Lua handler"); - let addr = ([127, 0, 0, 1], 3000).into(); - let server = Server::bind(&addr).executor(LocalExec).serve(MakeSvc(lua)); - - println!("Listening on http://{}", addr); - - // Create `LocalSet` to spawn !Send futures - let local = tokio::task::LocalSet::new(); - local.run_until(server).await.expect("cannot run server") -} - -struct MakeSvc(Rc<Lua>); - -impl Service<&AddrStream> for MakeSvc { - type Response = Svc; - type Error = hyper::Error; - type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>; - - fn poll_ready(&mut self, _: &mut Context) -> Poll<Result<(), Self::Error>> { - Poll::Ready(Ok(())) - } - - fn call(&mut self, stream: &AddrStream) -> Self::Future { - let lua = self.0.clone(); - let remote_addr = stream.remote_addr(); - Box::pin(async move { Ok(Svc(lua, remote_addr)) }) + let addr = "127.0.0.1:3000"; + + let local = LocalSet::new(); + let listener = TcpListener::bind(addr).await.unwrap(); + loop { + let (stream, peer_addr) = listener.accept().await.unwrap(); + let io = TokioIo::new(stream); + + let svc = Svc(lua.clone(), peer_addr); + local + .run_until(async move { + if let Err(err) = auto::Builder::new(LocalExec) + .http1() + .serve_connection(io, svc) + .await + { + println!("Error serving connection: {:?}", err); + } + }) + .await; } } |