http_mel/client.rs
1use crate::method::*;
2use crate::status::*;
3use async_ringbuf::AsyncHeapRb;
4use melodium_core::*;
5use melodium_macro::{check, mel_model, mel_treatment};
6use std::collections::HashMap;
7use std::sync::RwLock;
8use std::sync::{Arc, Weak};
9use std_mel::data::string_map::*;
10use trillium::HeaderName;
11use trillium::HeaderValue;
12use trillium::KnownHeaderName;
13use trillium_client::Url;
14use trillium_client::{Body, Client};
15
16pub const USER_AGENT: &str = concat!("http-mel/", env!("CARGO_PKG_VERSION"));
17
18/// HTTP client for general use
19///
20/// The HTTP client provides configuration for HTTP requests.
21///
22/// - `base_url`: The base URL for a client. All request URLs will be relative to this URL.
23/// - `tcp_no_delay`: TCP `NO_DELAY` field.
24/// - `headers`: Headers to add in requests made with this client.
25///
26/// The default headers are `Accept: */*` and `User-Agent: http-mel/<version>`
27#[mel_model(
28 param base_url Option<string> none
29 param tcp_no_delay bool true
30 param headers StringMap none
31 initialize initialization
32)]
33#[derive(Debug)]
34pub struct HttpClient {
35 model: Weak<HttpClientModel>,
36 client: RwLock<Option<Arc<Client>>>,
37}
38
39impl HttpClient {
40 fn new(model: Weak<HttpClientModel>) -> Self {
41 Self {
42 model,
43 client: RwLock::new(None),
44 }
45 }
46
47 fn initialization(&self) {
48 #[cfg(feature = "real")]
49 {
50 let model = self.model.upgrade().unwrap();
51
52 let config = trillium_rustls::RustlsConfig::default().with_tcp_config(
53 trillium_async_std::ClientConfig::new().with_nodelay(model.get_tcp_no_delay()),
54 );
55
56 let mut client = Client::new(config)
57 .with_default_pool()
58 .with_default_header(KnownHeaderName::UserAgent, USER_AGENT);
59 if let Some(base) = model.get_base_url() {
60 if let Ok(url) = Url::parse(&base) {
61 client = client.with_base(url);
62 }
63 }
64
65 *self.client.write().unwrap() = Some(Arc::new(client));
66 }
67 }
68
69 fn client(&self) -> Option<Arc<Client>> {
70 self.client.read().unwrap().clone()
71 }
72
73 fn invoke_source(&self, _source: &str, _params: HashMap<String, Value>) {}
74}
75
76/// Performs HTTP operation without data emission.
77///
78/// This treatment process HTTP request to the given `url`.
79/// - `method`: HTTP method used for the request.
80///
81/// - `url`: the URL to use for the request (combined with optional base from the client model), request starts as soon as the URL is transmitted.
82/// - `req_headers`: the headers to use for the request (combined with ones defined at client level).
83///
84/// - `status`: HTTP status response.
85/// - `res_headers`: the headers contained in the response.
86/// - `data`: data received as response, corresponding to the HTTP body.
87/// - `completed`: emitted when the incoming request finished successfully.
88/// - `failed`: emitted if the incoming request failed technically.
89/// - `error`: message containing error when request failed technically.
90/// - `finished`: emitted when the incoming request finished, regardless of state.
91#[mel_treatment(
92 model client HttpClient
93 input url Block<string>
94 input req_headers Block<StringMap>
95 output res_headers Block<StringMap>
96 output data Stream<byte>
97 output completed Block<void>
98 output failed Block<void>
99 output finished Block<void>
100 output error Block<string>
101 output status Block<HttpStatus>
102)]
103pub async fn request(method: HttpMethod) {
104 if let (Ok(url), Ok(req_headers)) = (
105 url.recv_one_as::<string>().await,
106 req_headers.recv_one_as::<Arc<StringMap>>().await,
107 ) {
108 if let Some(client) = HttpClientModel::into(client).inner().client() {
109 match client
110 .base()
111 .map(|base_url| base_url.join(&url))
112 .unwrap_or_else(|| Url::parse(&url))
113 {
114 Ok(url) => match {
115 let mut conn = client.build_conn(method.0, url);
116 for (name, content) in &req_headers.map {
117 let header_name = HeaderName::from(name.to_string());
118 if header_name.is_valid() {
119 let header_content = HeaderValue::from(content.clone());
120 if header_content.is_valid() {
121 conn.request_headers_mut()
122 .insert(header_name.to_owned(), header_content);
123 }
124 }
125 }
126 conn
127 }
128 .await
129 {
130 Ok(mut conn) => {
131 if let Some(recv_status) = conn.status() {
132 let _ = status
133 .send_one_as(Arc::new(HttpStatus(recv_status)) as Arc<dyn Data>)
134 .await;
135
136 let headers = conn
137 .response_headers()
138 .iter()
139 .filter_map(|(name, value)| {
140 value
141 .as_str()
142 .map(|value| (name.to_string(), value.to_string()))
143 })
144 .collect();
145
146 let _ =
147 res_headers
148 .send_one_as(
149 Arc::new(StringMap::new_with(headers)) as Arc<dyn Data>
150 )
151 .await;
152
153 status.close().await;
154 res_headers.close().await;
155
156 let data_buf = AsyncHeapRb::<u8>::new(2usize.pow(20));
157 let (prod, mut cons) = data_buf.split();
158
159 let response_body = conn.response_body();
160 let _ = futures::join!(
161 async {
162 let _ = async_std::io::copy(response_body, prod).await;
163 let _ = completed.send_one_as(()).await;
164 },
165 async {
166 loop {
167 let mut size = 2usize.pow(20);
168 let mut recv_data = vec![0; size];
169
170 match cons.pop_slice(&mut recv_data).await {
171 Ok(_) => {}
172 Err(written_size) => size = written_size,
173 }
174
175 recv_data.truncate(size);
176
177 check!(
178 data.send_many(TransmissionValue::Byte(
179 recv_data.into()
180 ))
181 .await
182 );
183 if cons.is_closed() {
184 break;
185 }
186 }
187 }
188 );
189 }
190 }
191 Err(err) => {
192 let _ = failed.send_one_as(()).await;
193 let _ = error.send_one_as(err.to_string()).await;
194 }
195 },
196 Err(err) => {
197 let _ = failed.send_one_as(()).await;
198 let _ = error.send_one_as(err.to_string()).await;
199 }
200 }
201 let _ = finished.send_one_as(()).await;
202 }
203 }
204}
205
206/// Performs HTTP operation with data emission.
207///
208/// This treatment process HTTP request to the given `url`.
209/// - `method`: HTTP method used for the request.
210///
211/// - `url`: the URL to use for the request (combined with optional base from the client model), request starts as soon as the URL is transmitted.
212/// - `req_headers`: the headers to use for the request (combined with ones defined at client level).
213/// - `body`: data to send as request body.
214///
215/// - `status`: HTTP status response.
216/// - `res_headers`: the headers contained in the response.
217/// - `data`: data received as response, corresponding to the HTTP body.
218/// - `completed`: emitted when the request finished successfully.
219/// - `failed`: emitted if the request failed technically.
220/// - `error`: message containing error when request failed technically.
221/// - `finished`: emitted when the request finished, regardless of state.
222#[mel_treatment(
223 model client HttpClient
224 input url Block<string>
225 input req_headers Block<StringMap>
226 input body Stream<byte>
227 output data Stream<byte>
228 output res_headers Block<StringMap>
229 output completed Block<void>
230 output failed Block<void>
231 output finished Block<void>
232 output error Block<string>
233 output status Block<HttpStatus>
234)]
235pub async fn request_with_body(method: HttpMethod) {
236 if let (Ok(url), Ok(req_headers)) = (
237 url.recv_one_as::<string>().await,
238 req_headers.recv_one_as::<Arc<StringMap>>().await,
239 ) {
240 if let Some(client) = HttpClientModel::into(client).inner().client() {
241 match client
242 .base()
243 .map(|base_url| base_url.join(&url))
244 .unwrap_or_else(|| Url::parse(&url))
245 {
246 Ok(url) => {
247 let in_body_buf = AsyncHeapRb::<u8>::new(2usize.pow(20));
248 let (mut in_prod, in_cons) = in_body_buf.split();
249
250 let conn_doing = async {
251 {
252 let mut conn = client.build_conn(method.0, url);
253
254 for (name, content) in &req_headers.map {
255 let header_name = HeaderName::from(name.to_string());
256 if header_name.is_valid() {
257 let header_content = HeaderValue::from(content.to_string());
258 if header_content.is_valid() {
259 conn.request_headers_mut()
260 .insert(header_name.to_owned(), header_content);
261 }
262 }
263 }
264 conn.with_body(Body::new_streaming(in_cons, None))
265 }
266 .await
267 };
268 let body_transmission = async {
269 while let Ok(body_data) = body
270 .recv_many()
271 .await
272 .map(|values| TryInto::<VecDeque<u8>>::try_into(values).unwrap())
273 {
274 if let Err(_) = in_prod.push_iter(body_data.into_iter()).await {
275 break;
276 }
277 }
278 in_prod.close();
279 };
280
281 match futures::join!(body_transmission, conn_doing) {
282 (_, Ok(mut conn)) => {
283 if let Some(recv_status) = conn.status() {
284 let _ = status
285 .send_one_as(Arc::new(HttpStatus(recv_status)) as Arc<dyn Data>)
286 .await;
287
288 let headers = conn
289 .response_headers()
290 .iter()
291 .filter_map(|(name, value)| {
292 value
293 .as_str()
294 .map(|value| (name.to_string(), value.to_string()))
295 })
296 .collect();
297 let _ = res_headers
298 .send_one_as(
299 Arc::new(StringMap::new_with(headers)) as Arc<dyn Data>
300 )
301 .await;
302
303 status.close().await;
304 res_headers.close().await;
305
306 let out_data_buf = AsyncHeapRb::<u8>::new(2usize.pow(20));
307 let (out_prod, mut out_cons) = out_data_buf.split();
308
309 let response_body = conn.response_body();
310 let _ = futures::join!(
311 async {
312 let _ = async_std::io::copy(response_body, out_prod).await;
313 let _ = completed.send_one_as(()).await;
314 },
315 async {
316 loop {
317 let mut size = 2usize.pow(20);
318 let mut recv_data = vec![0; size];
319 match out_cons.pop_slice(&mut recv_data).await {
320 Ok(_) => {}
321 Err(written_size) => size = written_size,
322 }
323
324 recv_data.truncate(size);
325
326 check!(
327 data.send_many(TransmissionValue::Byte(
328 recv_data.into()
329 ))
330 .await
331 );
332 if out_cons.is_closed() {
333 break;
334 }
335 }
336 }
337 );
338 }
339 }
340 (_, Err(err)) => {
341 let _ = failed.send_one_as(()).await;
342 let _ = error.send_one_as(err.to_string()).await;
343 }
344 }
345 }
346 Err(err) => {
347 let _ = failed.send_one_as(()).await;
348 let _ = error.send_one_as(err.to_string()).await;
349 }
350 }
351 let _ = finished.send_one_as(()).await;
352 }
353 }
354}