http_mel/
server.rs

1use crate::method::*;
2use crate::status::*;
3use async_ringbuf::{AsyncHeapRb, AsyncProducer, AsyncRb};
4use async_std::sync::{Arc as AsyncArc, Barrier as AsyncBarrier, RwLock as AsyncRwLock};
5use core::{fmt::Debug, mem::MaybeUninit};
6use melodium_core::{common::executive::ResultStatus, *};
7use melodium_macro::{mel_context, mel_model, mel_treatment};
8use net_mel::ip::*;
9use ringbuf::SharedRb;
10use routefinder::RouteSpec;
11use routefinder::Segment;
12use std::sync::Arc;
13use std::{
14    collections::HashMap,
15    sync::{RwLock, Weak},
16};
17use std_mel::data::string_map::*;
18use trillium::HeaderName;
19use trillium::HeaderValue;
20use trillium::KnownHeaderName;
21use trillium::{Body, Conn};
22use trillium::{Method, Status};
23use trillium_router::{Router, RouterConnExt};
24use uuid::Uuid;
25
26pub const SERVER: &str = concat!("http-mel/", env!("CARGO_PKG_VERSION"));
27
28/// Describes HTTP request data.
29///
30/// - `id`: identifier of connection, it is an arbitrary number that uniquely identifies a HTTP connection to a server.
31/// - `route`: the route used by the request.
32/// - `path`: the path called by the request.
33/// - `parameters`: the parameters from the route.
34/// - `method`: the HTTP method used by the request.
35#[mel_context]
36pub struct HttpRequest {
37    pub id: u128,
38    pub route: string,
39    pub path: string,
40    pub parameters: StringMap,
41    pub method: HttpMethod,
42}
43
44type AsyncProducerStatus =
45    AsyncProducer<Status, Arc<AsyncRb<Status, SharedRb<Status, Vec<MaybeUninit<Status>>>>>>;
46type AsyncProducerHeaders = AsyncProducer<
47    StringMap,
48    Arc<AsyncRb<StringMap, SharedRb<StringMap, Vec<MaybeUninit<StringMap>>>>>,
49>;
50type AsyncProducerOutgoing =
51    AsyncProducer<u8, Arc<AsyncRb<u8, SharedRb<u8, Vec<MaybeUninit<u8>>>>>>;
52
53/// A HTTP server for general use.
54///
55/// The HTTP server provides configuration for receiving and responding to HTTP incoming requests.
56/// - `host`: the network address to bind with.
57/// - `port`: the port to bind with.
58///
59/// `HttpServer` aims to be used with `connection` treatment.
60/// Every time a new HTTP request matching a configured route comes, a new track is created with `@HttpRequest` context.
61///
62/// ℹ️ If server binding fails, `failedBinding` is emitted.
63///
64/// ⚠️ Use `HttpServer` with `connection` treatment, as using `incoming` source and `outgoing` treatment directly should be done carefully.
65///
66#[mel_model(
67    param host Ip none
68    param port u16 none
69    source incoming (HttpRequest) (
70        param method HttpMethod none
71        param route string none
72    ) (
73        started Block<void>
74        headers Block<StringMap>
75        data Stream<byte>
76        finished Block<void>
77        completed Block<void>
78        failed Block<void>
79        error Block<string>
80    )
81    source failedBinding () () (
82        failed Block<void>
83        error Block<string>
84    )
85    continuous (continuous)
86    shutdown shutdown
87)]
88pub struct HttpServer {
89    model: Weak<HttpServerModel>,
90    launch_barrier: AsyncArc<AsyncBarrier>,
91    routes: RwLock<Vec<(Arc<HttpMethod>, String)>>,
92    status: AsyncArc<AsyncRwLock<HashMap<Uuid, AsyncProducerStatus>>>,
93    headers: AsyncArc<AsyncRwLock<HashMap<Uuid, AsyncProducerHeaders>>>,
94    outgoing: AsyncArc<AsyncRwLock<HashMap<Uuid, AsyncProducerOutgoing>>>,
95    #[cfg(feature = "real")]
96    shutdown: trillium_async_std::Stopper,
97}
98
99impl Debug for HttpServer {
100    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
101        #[cfg(feature = "real")]
102        {
103            f.debug_struct("HttpServer")
104                .field("model", &self.model)
105                .field("routes", &self.routes)
106                .field("shutdown", &self.shutdown)
107                .finish()
108        }
109
110        #[cfg(not(feature = "real"))]
111        {
112            f.debug_struct("HttpServer")
113                .field("model", &self.model)
114                .field("routes", &self.routes)
115                .finish()
116        }
117    }
118}
119
120impl HttpServer {
121    pub fn new(model: Weak<HttpServerModel>) -> Self {
122        Self {
123            model,
124            launch_barrier: AsyncArc::new(AsyncBarrier::new(2)),
125            routes: RwLock::new(Vec::new()),
126            status: AsyncArc::new(AsyncRwLock::new(HashMap::new())),
127            headers: AsyncArc::new(AsyncRwLock::new(HashMap::new())),
128            outgoing: AsyncArc::new(AsyncRwLock::new(HashMap::new())),
129            #[cfg(feature = "real")]
130            shutdown: trillium_async_std::Stopper::new(),
131        }
132    }
133
134    pub fn statuses(&self) -> AsyncArc<AsyncRwLock<HashMap<Uuid, AsyncProducerStatus>>> {
135        AsyncArc::clone(&self.status)
136    }
137
138    pub fn headers(&self) -> AsyncArc<AsyncRwLock<HashMap<Uuid, AsyncProducerHeaders>>> {
139        AsyncArc::clone(&self.headers)
140    }
141
142    pub fn outgoing(&self) -> AsyncArc<AsyncRwLock<HashMap<Uuid, AsyncProducerOutgoing>>> {
143        AsyncArc::clone(&self.outgoing)
144    }
145
146    #[cfg(feature = "real")]
147    async fn continuous(&self) {
148        let model = self.model.upgrade().unwrap();
149
150        self.launch_barrier.wait().await;
151
152        let routes = self.routes.read().unwrap().clone();
153
154        let status = self.status.clone();
155        let headers = self.headers.clone();
156        let outgoing = self.outgoing.clone();
157
158        let mut router = Router::new();
159        for (method, route) in routes {
160            let route = match RouteSpec::try_from(route.as_str()) {
161                Ok(route) => route,
162                Err(_) => continue,
163            };
164
165            let handler = {
166                let route = Arc::new(route.clone());
167                let status = Arc::clone(&status);
168                let headers = Arc::clone(&headers);
169                let outgoing = Arc::clone(&outgoing);
170                let model = Arc::clone(&model);
171                let method = Arc::clone(&method);
172
173                move |mut conn: Conn| {
174                    let route = Arc::clone(&route);
175                    let status = Arc::clone(&status);
176                    let headers = Arc::clone(&headers);
177                    let outgoing = Arc::clone(&outgoing);
178                    let model = Arc::clone(&model);
179                    let method = Arc::clone(&method);
180
181                    async move {
182                        let id = Uuid::new_v4();
183                        let http_request = HttpRequest {
184                            id: id.as_u128(),
185                            route: conn.route().map(|r| r.to_string()).unwrap_or_default(),
186                            path: conn.path().to_string(),
187                            parameters: StringMap::new_with(
188                                route
189                                    .segments()
190                                    .iter()
191                                    .filter_map(|seg| {
192                                        if let Segment::Param(param) = seg {
193                                            conn.param(param)
194                                                .map(|v| (param.to_string(), v.to_string()))
195                                        } else {
196                                            None
197                                        }
198                                    })
199                                    .collect(),
200                            ),
201                            method: (*method).clone(),
202                        };
203
204                        let params = {
205                            let mut params = HashMap::new();
206                            params.insert(
207                                "method".to_string(),
208                                Value::Data(Arc::clone(&method) as Arc<dyn Data>),
209                            );
210                            params.insert("route".to_string(), route.to_string().into());
211                            params
212                        };
213
214                        let status_buf = AsyncHeapRb::<Status>::new(1);
215                        let (status_prod, mut status_cons) = status_buf.split();
216                        let headers_buf = AsyncHeapRb::<StringMap>::new(1);
217                        let (headers_prod, mut headers_cons) = headers_buf.split();
218                        let outgoing_buf = AsyncHeapRb::<u8>::new(2usize.pow(20));
219                        let (prod, cons) = outgoing_buf.split();
220
221                        status.write().await.insert(id, status_prod);
222                        headers.write().await.insert(id, headers_prod);
223                        outgoing.write().await.insert(id, prod);
224
225                        let incoming_headers = conn
226                            .request_headers()
227                            .iter()
228                            .filter_map(|(name, value)| {
229                                value
230                                    .as_str()
231                                    .map(|value| (name.to_string(), value.to_string()))
232                            })
233                            .collect();
234
235                        // For now the reading of request is "one-shot", not allowing effective streaming of very large incoming requests.
236                        let body = conn.request_body().await;
237                        let (content, occured_failure) = match body.read_bytes().await {
238                            Ok(content) => (content, None),
239                            Err(err) => (Vec::new(), Some(err.to_string())),
240                        };
241
242                        model
243                            .new_incoming(
244                                None,
245                                http_request,
246                                &params,
247                                Some(Box::new(move |mut outputs| {
248                                    let started = outputs.get("started");
249                                    let headers = outputs.get("headers");
250                                    let data = outputs.get("data");
251                                    let completed = outputs.get("completed");
252                                    let failed = outputs.get("failed");
253                                    let finished = outputs.get("finished");
254                                    let error = outputs.get("error");
255
256                                    vec![Box::new(Box::pin(async move {
257                                        if let Some(occured_failure) = occured_failure {
258                                            let _ = failed.send_one_as(()).await;
259                                            let _ = error.send_one_as(occured_failure).await;
260                                        } else {
261                                            let _ = started.send_one_as(()).await;
262                                            started.close().await;
263                                            let _ = headers
264                                                .send_one_as(Arc::new(StringMap::new_with(
265                                                    incoming_headers,
266                                                ))
267                                                    as Arc<dyn Data>)
268                                                .await;
269                                            headers.close().await;
270                                            let _ = data
271                                                .send_many(TransmissionValue::Byte(content.into()))
272                                                .await;
273                                            let _ = completed.send_one_as(()).await;
274                                        }
275                                        let _ = finished.send_one_as(()).await;
276
277                                        headers.close().await;
278                                        data.close().await;
279                                        completed.close().await;
280                                        finished.close().await;
281                                        failed.close().await;
282                                        error.close().await;
283                                        ResultStatus::Ok
284                                    }))]
285                                })),
286                            )
287                            .await;
288
289                        if let (Some(status), Some(headers)) =
290                            futures::join!(status_cons.pop(), headers_cons.pop())
291                        {
292                            conn.set_status(status);
293                            conn.response_headers_mut()
294                                .insert(KnownHeaderName::Server, SERVER);
295
296                            for (name, content) in &headers.map {
297                                let header_name = HeaderName::from(name.to_string());
298                                if header_name.is_valid() {
299                                    let header_content = HeaderValue::from(content.clone());
300                                    if header_content.is_valid() {
301                                        conn.response_headers_mut()
302                                            .insert(header_name.to_owned(), header_content);
303                                    }
304                                }
305                            }
306
307                            conn.set_body(Body::new_streaming(cons, None));
308                        } else {
309                            conn.set_status(Status::InternalServerError);
310                        }
311
312                        conn.halt()
313                    }
314                }
315            };
316
317            match method.0 {
318                Method::Delete => router = router.delete(route, handler),
319                Method::Get => router = router.get(route, handler),
320                Method::Patch => router = router.patch(route, handler),
321                Method::Post => router = router.post(route, handler),
322                Method::Put => router = router.put(route, handler),
323                _ => {}
324            }
325        }
326
327        match async_std::net::TcpListener::bind((model.get_host().0, model.get_port())).await {
328            Ok(listener) => {
329                trillium_async_std::config()
330                    .without_signals()
331                    .with_stopper(self.shutdown.clone())
332                    .with_prebound_server(listener)
333                    .run_async(router)
334                    .await
335            }
336            Err(err) => {
337                model
338                    .new_failedBinding(
339                        None,
340                        &HashMap::new(),
341                        Some(Box::new(move |mut outputs| {
342                            let error = outputs.get("error");
343                            let failed = outputs.get("failed");
344                            vec![Box::new(Box::pin(async move {
345                                let _ = failed.send_one_as(()).await;
346                                let _ = error.send_one_as(err.to_string()).await;
347                                failed.close().await;
348                                error.close().await;
349                                ResultStatus::Ok
350                            }))]
351                        })),
352                    )
353                    .await
354            }
355        }
356    }
357
358    #[cfg(not(feature = "real"))]
359    async fn continuous(&self) {}
360
361    fn invoke_source(&self, source: &str, params: HashMap<String, Value>) {
362        match source {
363            "incoming" => {
364                let method = params
365                    .get("method")
366                    .unwrap()
367                    .clone()
368                    .try_data::<Arc<HttpMethod>>()
369                    .unwrap();
370                let route = params
371                    .get("route")
372                    .unwrap()
373                    .clone()
374                    .try_data::<String>()
375                    .unwrap();
376
377                self.routes.write().unwrap().push((method, route));
378            }
379            _ => {}
380        }
381    }
382
383    fn shutdown(&self) {
384        #[cfg(feature = "real")]
385        self.shutdown.stop();
386    }
387}
388
389/// Start listening for incoming HTTP connections.
390///
391/// Blocks until `trigger` is received, then releases the HTTP server launch barrier so
392/// that the server begins accepting connections.
393///
394/// ℹ️ Use this treatment together with `connection`: `start` unblocks the server, while
395/// each `connection` track handles an individual request.
396///
397/// ```mermaid
398/// graph LR
399///     T("start()")
400///     B["〈🟦〉"] -->|trigger| T
401///     style B fill:#ffffff,stroke:#ffffff
402/// ```
403#[mel_treatment(
404    model http_server HttpServer
405    input trigger Block<void>
406)]
407pub async fn start() {
408    let model = HttpServerModel::into(http_server);
409    let http_server = model.inner();
410
411    if let Ok(_) = trigger.recv_one().await {
412        http_server.launch_barrier.wait().await;
413    }
414}
415
416/// Send an HTTP response for a specific connection.
417///
418/// Low-level counterpart to `connection`, sending `status`, `headers`, and streaming the body
419/// `data` back to the client identified by `id`.
420///
421/// `status` and `headers` must both arrive before any body data is forwarded; if either is
422/// missing the response is not sent.
423///
424/// ⚠️ Use `connection` rather than `outgoing` directly unless you have a specific reason to
425/// manage connection IDs manually.
426///
427/// ```mermaid
428/// graph LR
429///     T("outgoing()")
430///     ST["〈🟦〉"] -->|status| T
431///     H["〈🟨〉"] -->|headers| T
432///     D["🟩 …"] -->|data| T
433///     style ST fill:#ffffff,stroke:#ffffff
434///     style H fill:#ffffff,stroke:#ffffff
435///     style D fill:#ffffff,stroke:#ffffff
436/// ```
437#[mel_treatment(
438    input status Block<HttpStatus>
439    input headers Block<StringMap>
440    input data Stream<byte>
441    model http_server HttpServer
442)]
443pub async fn outgoing(id: u128) {
444    let id = Uuid::from_u128(id);
445    let model = HttpServerModel::into(http_server);
446    let http_server = model.inner();
447
448    let out_status;
449    let out_headers;
450    let output;
451    {
452        let statuses = http_server.statuses();
453        let mut lock = statuses.write().await;
454        out_status = lock.remove(&id);
455    }
456    {
457        let headers = http_server.headers();
458        let mut lock = headers.write().await;
459        out_headers = lock.remove(&id);
460    }
461    {
462        let outputs = http_server.outgoing();
463        let mut lock = outputs.write().await;
464        output = lock.remove(&id);
465    }
466    if let (Some(mut out_status), Some(mut out_headers), Some(mut output)) =
467        (out_status, out_headers, output)
468    {
469        if let Ok((status, headers)) = futures::try_join!(
470            status.recv_one_as::<Arc<HttpStatus>>(),
471            headers.recv_one_as::<Arc<StringMap>>(),
472        ) {
473            match futures::join!(
474                out_status.push(status.0),
475                out_headers.push(Arc::unwrap_or_clone(headers))
476            ) {
477                (Ok(_), Ok(_)) => {
478                    while let (Ok(data), false) =
479                        (data.recv_many_as::<byte>().await, output.is_closed())
480                    {
481                        match output.push_iter(data.into_iter()).await {
482                            Ok(_) => {}
483                            Err(_) => break,
484                        }
485                    }
486                }
487                (_, _) => {}
488            }
489        }
490    }
491}
http_mel/server.rs

http_mel/
server.rs
