1use crate::engine::ecs::component::HttpClientComponent;
2use crate::engine::ecs::{ComponentId, EventSignal, SignalEmitter, World};
3use reqwest::blocking::Client;
4use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
5use std::collections::HashMap;
6use std::sync::mpsc::{self, Receiver, Sender};
7use std::time::Duration;
8
9#[derive(Debug)]
10struct HttpClientRuntime {
11 client: Client,
12}
13
14#[derive(Debug)]
15enum HttpClientCompletion {
16 Response {
17 component_id: ComponentId,
18 request_id: u64,
19 status: u16,
20 headers: Vec<(String, String)>,
21 body_text: String,
22 url: String,
23 },
24 Error {
25 component_id: ComponentId,
26 request_id: u64,
27 phase: String,
28 message: String,
29 url: String,
30 },
31}
32
33#[derive(Debug)]
34pub struct HttpClientSystem {
35 runtimes: HashMap<ComponentId, HttpClientRuntime>,
36 completion_tx: Sender<HttpClientCompletion>,
37 completion_rx: Receiver<HttpClientCompletion>,
38 next_request_id: u64,
39}
40
41impl Default for HttpClientSystem {
42 fn default() -> Self {
43 let (completion_tx, completion_rx) = mpsc::channel();
44 Self {
45 runtimes: HashMap::new(),
46 completion_tx,
47 completion_rx,
48 next_request_id: 1,
49 }
50 }
51}
52
53impl HttpClientSystem {
54 pub fn register_component(
55 &mut self,
56 world: &World,
57 emit: &mut dyn SignalEmitter,
58 component_id: ComponentId,
59 ) {
60 let Some(component) = world.get_component_by_id_as::<HttpClientComponent>(component_id)
61 else {
62 return;
63 };
64 if !component.enabled {
65 self.runtimes.remove(&component_id);
66 return;
67 }
68
69 let mut builder = Client::builder();
70 if let Some(timeout_ms) = component.timeout_ms {
71 builder = builder.timeout(Duration::from_millis(timeout_ms));
72 }
73
74 match builder.build() {
75 Ok(client) => {
76 self.runtimes
77 .insert(component_id, HttpClientRuntime { client });
78 }
79 Err(error) => emit.push_event(
80 component_id,
81 EventSignal::HttpError {
82 request_id: None,
83 phase: "register".to_string(),
84 message: error.to_string(),
85 url: None,
86 bind_addr: None,
87 },
88 ),
89 }
90 }
91
92 pub fn remove_component(&mut self, component_id: ComponentId) {
93 self.runtimes.remove(&component_id);
94 }
95
96 pub fn issue_request(
97 &mut self,
98 component_id: ComponentId,
99 method: String,
100 url: String,
101 headers: Vec<(String, String)>,
102 body_text: Option<String>,
103 ) {
104 let Some(runtime) = self.runtimes.get(&component_id) else {
105 return;
106 };
107
108 let request_id = self.next_request_id;
109 self.next_request_id += 1;
110
111 let tx = self.completion_tx.clone();
112 let client = runtime.client.clone();
113 std::thread::spawn(move || {
114 let method_parsed = match reqwest::Method::from_bytes(method.as_bytes()) {
115 Ok(method) => method,
116 Err(error) => {
117 let _ = tx.send(HttpClientCompletion::Error {
118 component_id,
119 request_id,
120 phase: "build".to_string(),
121 message: error.to_string(),
122 url,
123 });
124 return;
125 }
126 };
127
128 let mut header_map = HeaderMap::new();
129 for (name, value) in &headers {
130 let Ok(name) = HeaderName::try_from(name.as_str()) else {
131 let _ = tx.send(HttpClientCompletion::Error {
132 component_id,
133 request_id,
134 phase: "build".to_string(),
135 message: format!("invalid header name: {name}"),
136 url,
137 });
138 return;
139 };
140 let Ok(value) = HeaderValue::from_str(value) else {
141 let _ = tx.send(HttpClientCompletion::Error {
142 component_id,
143 request_id,
144 phase: "build".to_string(),
145 message: format!("invalid header value for {}", name.as_str()),
146 url,
147 });
148 return;
149 };
150 header_map.append(name, value);
151 }
152
153 let mut request = client.request(method_parsed, &url).headers(header_map);
154 if let Some(body_text) = body_text {
155 request = request.body(body_text);
156 }
157
158 match request.send() {
159 Ok(response) => {
160 let status = response.status().as_u16();
161 let response_url = response.url().to_string();
162 let headers = response
163 .headers()
164 .iter()
165 .map(|(name, value)| {
166 (
167 name.as_str().to_string(),
168 value.to_str().unwrap_or_default().to_string(),
169 )
170 })
171 .collect();
172 match response.text() {
173 Ok(body_text) => {
174 let _ = tx.send(HttpClientCompletion::Response {
175 component_id,
176 request_id,
177 status,
178 headers,
179 body_text,
180 url: response_url,
181 });
182 }
183 Err(error) => {
184 let _ = tx.send(HttpClientCompletion::Error {
185 component_id,
186 request_id,
187 phase: "read_body".to_string(),
188 message: error.to_string(),
189 url: response_url,
190 });
191 }
192 }
193 }
194 Err(error) => {
195 let _ = tx.send(HttpClientCompletion::Error {
196 component_id,
197 request_id,
198 phase: "request".to_string(),
199 message: error.to_string(),
200 url,
201 });
202 }
203 }
204 });
205 }
206
207 pub fn drain_completions(&mut self, emit: &mut dyn SignalEmitter) {
208 while let Ok(completion) = self.completion_rx.try_recv() {
209 match completion {
210 HttpClientCompletion::Response {
211 component_id,
212 request_id,
213 status,
214 headers,
215 body_text,
216 url,
217 } => {
218 if !self.runtimes.contains_key(&component_id) {
219 continue;
220 }
221 emit.push_event(
222 component_id,
223 EventSignal::HttpResponse {
224 request_id,
225 status,
226 ok: (200..300).contains(&status),
227 headers,
228 body_text,
229 url,
230 },
231 );
232 }
233 HttpClientCompletion::Error {
234 component_id,
235 request_id,
236 phase,
237 message,
238 url,
239 } => {
240 if !self.runtimes.contains_key(&component_id) {
241 continue;
242 }
243 emit.push_event(
244 component_id,
245 EventSignal::HttpError {
246 request_id: Some(request_id),
247 phase,
248 message,
249 url: Some(url),
250 bind_addr: None,
251 },
252 );
253 }
254 }
255 }
256 }
257}
258
259#[cfg(test)]
260mod tests {
261 use crate::engine::ecs::component::{
262 HttpClientComponent, HttpServerComponent, TransformComponent,
263 };
264 use crate::engine::ecs::{CommandQueue, EventSignal, SignalKind, SystemWorld, World};
265 use crate::engine::graphics::{RenderAssets, VisualWorld};
266 use std::sync::mpsc;
267 use std::time::{Duration, Instant};
268
269 #[test]
270 fn client_request_surfaces_http_response_event() {
271 let mut world = World::default();
272 let mut visuals = VisualWorld::default();
273 let mut render_assets = RenderAssets::new();
274 let mut systems = SystemWorld::default();
275 let mut queue = CommandQueue::new();
276 let (tx, rx) = mpsc::channel();
277
278 let server_root = world.add_component(TransformComponent::new());
279 let server = world.add_component(HttpServerComponent::bind("127.0.0.1:18081"));
280 let _ = world.add_child(server_root, server);
281 let client_root = world.add_component(TransformComponent::new());
282 let client = world.add_component(HttpClientComponent::new());
283 let _ = world.add_child(client_root, client);
284
285 systems.rx.add_handler_closure(
286 SignalKind::HttpRequest,
287 server,
288 move |_world, emit, signal| {
289 let Some(EventSignal::HttpRequest { request_id, .. }) = signal.event.as_ref()
290 else {
291 return;
292 };
293 emit.push_intent_now(
294 signal.scope,
295 crate::engine::ecs::IntentValue::HttpServerReply {
296 component_id: signal.scope,
297 request_id: *request_id,
298 status: 200,
299 headers: vec![],
300 body_text: "ok".to_string(),
301 },
302 );
303 },
304 );
305 systems.rx.add_handler_closure(
306 SignalKind::HttpResponse,
307 client,
308 move |_world, _emit, signal| {
309 let Some(EventSignal::HttpResponse { body_text, .. }) = signal.event.as_ref()
310 else {
311 return;
312 };
313 let _ = tx.send(body_text.clone());
314 },
315 );
316
317 world.init_component_tree(server_root, &mut queue);
318 world.init_component_tree(client_root, &mut queue);
319 systems.process_commands(&mut world, &mut visuals, &mut render_assets, &mut queue);
320
321 systems.http_client.issue_request(
322 client,
323 "GET".to_string(),
324 "http://127.0.0.1:18081/".to_string(),
325 vec![],
326 None,
327 );
328
329 let deadline = Instant::now() + Duration::from_secs(3);
330 while Instant::now() < deadline {
331 let _ = systems.process_signals(
332 &mut world,
333 &mut visuals,
334 &mut render_assets,
335 &mut queue,
336 100_000,
337 );
338 systems.process_commands(&mut world, &mut visuals, &mut render_assets, &mut queue);
339 if let Ok(body) = rx.try_recv() {
340 assert_eq!(body, "ok");
341 return;
342 }
343 std::thread::sleep(Duration::from_millis(10));
344 }
345
346 panic!("timed out waiting for HttpResponse");
347 }
348}