update hyper to v1, and add shell command example (#384)

Co-authored-by: Alex Orlenko <[email protected]>

Author: getong

Cargo.toml

@@ -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"] }

README.md

@@ -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).

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()?;

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,16 +45,28 @@ 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())
                 }
                 Err(err) => {
                     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;
     }
 }