Skip to content

Commit 94cef89

Browse files
committed
Update hyper examples
1 parent 849206e commit 94cef89

File tree

4 files changed

+109
-101
lines changed

4 files changed

+109
-101
lines changed

Cargo.toml

+2-3
Original file line numberDiff line numberDiff line change
@@ -63,10 +63,9 @@ libloading = { version = "0.8", optional = true }
6363
[dev-dependencies]
6464
trybuild = "1.0"
6565
futures = "0.3.5"
66-
hyper = { version = "1", features = ["client", "server"] }
67-
hyper-util = { version = "0.1.3", features = ["server", "client", "client-legacy", "tokio", "http1"] }
66+
hyper = { version = "1.2", features = ["full"] }
67+
hyper-util = { version = "0.1.3", features = ["full"] }
6868
http-body-util = "0.1.1"
69-
bytes = "1.5.0"
7069
reqwest = { version = "0.12", features = ["json"] }
7170
tokio = { version = "1.0", features = ["macros", "rt", "time"] }
7271
serde = { version = "1.0", features = ["derive"] }

README.md

+9-6
Original file line numberDiff line numberDiff line change
@@ -85,14 +85,17 @@ This works using Lua [coroutines](https://www.lua.org/manual/5.3/manual.html#2.6
8585
- [TCP Server](examples/async_tcp_server.rs)
8686

8787

88-
**shell command example**:
89-
``` shell
90-
# async http client
91-
cargo run --example async_http_client --features="async macros lua54"
88+
**shell command examples**:
89+
```shell
90+
# async http client (hyper)
91+
foo@bar:~$ cargo run --example async_http_client --features=lua54,async,macros
92+
93+
# async http client (reqwest)
94+
foo@bar:~$ cargo run --example async_http_reqwest --features=lua54,async,macros,serialize
9295

9396
# async http server
94-
cargo run --example async_http_server --features="async macros lua54"
95-
curl http://localhost:3000
97+
foo@bar:~$ cargo run --example async_http_server --features=lua54,async,macros
98+
foo@bar:~$ curl -v http://localhost:3000
9699
```
97100

98101
### Serialization (serde) support

examples/async_http_client.rs

+19-34
Original file line numberDiff line numberDiff line change
@@ -1,38 +1,24 @@
1-
use bytes::Bytes;
2-
use http_body_util::BodyExt;
3-
use http_body_util::Empty;
1+
use std::collections::HashMap;
2+
3+
use http_body_util::BodyExt as _;
44
use hyper::body::Incoming;
55
use hyper_util::client::legacy::Client as HyperClient;
6-
use std::collections::HashMap;
6+
use hyper_util::rt::TokioExecutor;
77

88
use mlua::{chunk, ExternalResult, Lua, Result, UserData, UserDataMethods};
99

1010
struct BodyReader(Incoming);
1111

1212
impl UserData for BodyReader {
1313
fn add_methods<'lua, M: UserDataMethods<'lua, Self>>(methods: &mut M) {
14+
// Every call returns a next chunk
1415
methods.add_async_method_mut("read", |lua, reader, ()| async move {
15-
let mut summarize = Vec::new(); // Create a vector to accumulate the bytes
16-
17-
loop {
18-
match reader.0.frame().await {
19-
Some(Ok(bytes)) => {
20-
if let Ok(data) = bytes.into_data() {
21-
summarize.extend(data); // Append the bytes to the summarize variable
22-
}
23-
}
24-
Some(Err(_)) => break, // Break on error
25-
None => break, // Break if no more frames
16+
if let Some(bytes) = reader.0.frame().await {
17+
if let Some(bytes) = bytes.into_lua_err()?.data_ref() {
18+
return Some(lua.create_string(&bytes)).transpose();
2619
}
2720
}
28-
29-
if !summarize.is_empty() {
30-
// If summarize has collected data, return it as a Lua string
31-
Ok(Some(lua.create_string(&summarize)?))
32-
} else {
33-
// Return None if no data was collected
34-
Ok(None)
35-
}
21+
Ok(None)
3622
});
3723
}
3824
}
@@ -42,8 +28,7 @@ async fn main() -> Result<()> {
4228
let lua = Lua::new();
4329

4430
let fetch_url = lua.create_async_function(|lua, uri: String| async move {
45-
let client =
46-
HyperClient::builder(hyper_util::rt::TokioExecutor::new()).build_http::<Empty<Bytes>>();
31+
let client = HyperClient::builder(TokioExecutor::new()).build_http::<String>();
4732
let uri = uri.parse().into_lua_err()?;
4833
let resp = client.get(uri).await.into_lua_err()?;
4934

@@ -66,19 +51,19 @@ async fn main() -> Result<()> {
6651

6752
let f = lua
6853
.load(chunk! {
69-
local res = $fetch_url(...)
54+
local res = $fetch_url(...)
7055
print("status: "..res.status)
7156
for key, vals in pairs(res.headers) do
72-
for _, val in ipairs(vals) do
73-
print(key..": "..val)
74-
end
57+
for _, val in ipairs(vals) do
58+
print(key..": "..val)
59+
end
7560
end
7661
repeat
77-
local body = res.body:read()
78-
if body then
79-
print(body)
80-
end
81-
until not body
62+
local chunk = res.body:read()
63+
if chunk then
64+
print(chunk)
65+
end
66+
until not chunk
8267
})
8368
.into_function()?;
8469

examples/async_http_server.rs

+79-58
Original file line numberDiff line numberDiff line change
@@ -1,35 +1,63 @@
1-
use bytes::Bytes;
2-
use http_body_util::{combinators::BoxBody, BodyExt, Empty, Full};
3-
use hyper::{body::Incoming, service::Service, Request, Response};
4-
use hyper_util::{rt::TokioIo, server::conn::auto};
1+
use std::convert::Infallible;
2+
use std::future::Future;
3+
use std::net::SocketAddr;
4+
use std::rc::Rc;
5+
6+
use futures::future::LocalBoxFuture;
7+
use http_body_util::{combinators::BoxBody, BodyExt as _, Empty, Full};
8+
use hyper::body::{Bytes, Incoming};
9+
use hyper::{Request, Response};
10+
use hyper_util::rt::TokioIo;
11+
use hyper_util::server::conn::auto::Builder as ServerConnBuilder;
12+
use tokio::net::TcpListener;
13+
use tokio::task::LocalSet;
14+
515
use mlua::{
6-
chunk, Error as LuaError, Function, Lua, String as LuaString, Table, UserData, UserDataMethods,
16+
chunk, Error as LuaError, Function, Lua, RegistryKey, String as LuaString, Table, UserData,
17+
UserDataMethods,
718
};
8-
use std::{future::Future, net::SocketAddr, pin::Pin, rc::Rc};
9-
use tokio::{net::TcpListener, task::LocalSet};
1019

20+
/// Wrapper around incoming request that implements UserData
1121
struct LuaRequest(SocketAddr, Request<Incoming>);
1222

1323
impl UserData for LuaRequest {
1424
fn add_methods<'lua, M: UserDataMethods<'lua, Self>>(methods: &mut M) {
15-
methods.add_method("remote_addr", |_lua, req, ()| Ok((req.0).to_string()));
16-
methods.add_method("method", |_lua, req, ()| Ok((req.1).method().to_string()));
25+
methods.add_method("remote_addr", |_, req, ()| Ok((req.0).to_string()));
26+
methods.add_method("method", |_, req, ()| Ok((req.1).method().to_string()));
27+
methods.add_method("path", |_, req, ()| Ok(req.1.uri().path().to_string()));
1728
}
1829
}
1930

20-
pub struct Svc(Rc<Lua>, SocketAddr);
31+
/// Service that handles incoming requests
32+
#[derive(Clone)]
33+
pub struct Svc {
34+
lua: Rc<Lua>,
35+
handler: Rc<RegistryKey>,
36+
peer_addr: SocketAddr,
37+
}
38+
39+
impl Svc {
40+
pub fn new(lua: Rc<Lua>, handler: Rc<RegistryKey>, peer_addr: SocketAddr) -> Self {
41+
Self {
42+
lua,
43+
handler,
44+
peer_addr,
45+
}
46+
}
47+
}
2148

22-
impl Service<Request<Incoming>> for Svc {
23-
type Response = Response<BoxBody<Bytes, hyper::Error>>;
49+
impl hyper::service::Service<Request<Incoming>> for Svc {
50+
type Response = Response<BoxBody<Bytes, Infallible>>;
2451
type Error = LuaError;
25-
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
52+
type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;
2653

2754
fn call(&self, req: Request<Incoming>) -> Self::Future {
2855
// If handler returns an error then generate 5xx response
29-
let lua = self.0.clone();
30-
let lua_req = LuaRequest(self.1, req);
56+
let lua = self.lua.clone();
57+
let handler_key = self.handler.clone();
58+
let lua_req = LuaRequest(self.peer_addr, req);
3159
Box::pin(async move {
32-
let handler: Function = lua.named_registry_value("http_handler")?;
60+
let handler: Function = lua.registry_value(&handler_key)?;
3361
match handler.call_async::<_, Table>(lua_req).await {
3462
Ok(lua_resp) => {
3563
let status = lua_resp.get::<_, Option<u16>>("status")?.unwrap_or(200);
@@ -43,30 +71,19 @@ impl Service<Request<Incoming>> for Svc {
4371
}
4472
}
4573

74+
// Set body
4675
let body = lua_resp
4776
.get::<_, Option<LuaString>>("body")?
48-
.map(|b| {
49-
Full::new(Bytes::copy_from_slice(b.clone().as_bytes()))
50-
.map_err(|never| match never {})
51-
.boxed()
52-
})
53-
.unwrap_or_else(|| {
54-
Empty::<Bytes>::new()
55-
.map_err(|never| match never {})
56-
.boxed()
57-
});
77+
.map(|b| Full::new(Bytes::copy_from_slice(b.as_bytes())).boxed())
78+
.unwrap_or_else(|| Empty::<Bytes>::new().boxed());
5879

5980
Ok(resp.body(body).unwrap())
6081
}
6182
Err(err) => {
6283
eprintln!("{}", err);
6384
Ok(Response::builder()
6485
.status(500)
65-
.body(
66-
Full::new(Bytes::from("Internal Server Error".as_bytes()))
67-
.map_err(|never| match never {})
68-
.boxed(),
69-
)
86+
.body(Full::new(Bytes::from("Internal Server Error")).boxed())
7087
.unwrap())
7188
}
7289
}
@@ -79,43 +96,47 @@ async fn main() {
7996
let lua = Rc::new(Lua::new());
8097

8198
// Create Lua handler function
82-
let handler: Function = lua
99+
let handler: RegistryKey = lua
83100
.load(chunk! {
84-
function(req)
85-
return {
86-
status = 200,
87-
headers = {
88-
["X-Req-Method"] = req:method(),
89-
["X-Remote-Addr"] = req:remote_addr(),
90-
},
91-
body = "Hello from Lua!\n"
92-
}
93-
end
101+
function(req)
102+
return {
103+
status = 200,
104+
headers = {
105+
["X-Req-Method"] = req:method(),
106+
["X-Req-Path"] = req:path(),
107+
["X-Remote-Addr"] = req:remote_addr(),
108+
},
109+
body = "Hello from Lua!\n"
110+
}
111+
end
94112
})
95113
.eval()
96-
.expect("cannot create Lua handler");
114+
.expect("Failed to create Lua handler");
115+
let handler = Rc::new(handler);
97116

98-
// Store it in the Registry
99-
lua.set_named_registry_value("http_handler", handler)
100-
.expect("cannot store Lua handler");
101-
102-
let addr = "127.0.0.1:3000";
117+
let listen_addr = "127.0.0.1:3000";
118+
let listener = TcpListener::bind(listen_addr).await.unwrap();
119+
println!("Listening on http://{listen_addr}");
103120

104121
let local = LocalSet::new();
105-
let listener = TcpListener::bind(addr).await.unwrap();
106122
loop {
107-
let (stream, peer_addr) = listener.accept().await.unwrap();
108-
let io = TokioIo::new(stream);
123+
let (stream, peer_addr) = match listener.accept().await {
124+
Ok(x) => x,
125+
Err(err) => {
126+
eprintln!("Failed to accept connection: {err}");
127+
continue;
128+
}
129+
};
109130

110-
let svc = Svc(lua.clone(), peer_addr);
131+
let svc = Svc::new(lua.clone(), handler.clone(), peer_addr);
111132
local
112133
.run_until(async move {
113-
if let Err(err) = auto::Builder::new(LocalExec)
134+
let result = ServerConnBuilder::new(LocalExec)
114135
.http1()
115-
.serve_connection(io, svc)
116-
.await
117-
{
118-
println!("Error serving connection: {:?}", err);
136+
.serve_connection(TokioIo::new(stream), svc)
137+
.await;
138+
if let Err(err) = result {
139+
eprintln!("Error serving connection: {err:?}");
119140
}
120141
})
121142
.await;
@@ -127,7 +148,7 @@ struct LocalExec;
127148

128149
impl<F> hyper::rt::Executor<F> for LocalExec
129150
where
130-
F: std::future::Future + 'static, // not requiring `Send`
151+
F: Future + 'static, // not requiring `Send`
131152
{
132153
fn execute(&self, fut: F) {
133154
tokio::task::spawn_local(fut);

0 commit comments

Comments
 (0)