zng_task/http.rs
1#![cfg(feature = "http")]
2
3//! HTTP client.
4//!
5//! This module provides an HTTP client API that is backend agnostic. By default it uses the system `curl` command
6//! line utility with a simple cache, this can be replaced by implementing [`HttpClient`] and [`HttpCache`].
7//!
8
9mod cache;
10mod ctx;
11mod curl;
12mod file_cache;
13mod stream_decoder;
14mod util;
15
16pub use cache::{CacheKey, CacheMode, CachePolicy};
17pub use ctx::{HttpCache, HttpClient, http_cache, http_client, set_http_cache, set_http_client, set_request_default};
18pub use curl::CurlProcessClient;
19pub use file_cache::FileSystemCache;
20
21/// Any error sending request or receiving response.
22///
23/// # HTTP Error
24///
25/// In functions that return `Result<Response, Error>` successfully received HTTP errors are not converted to `Err`. Use
26/// [`Response::error`] to get a [`HttpError`] if the response represents an error.
27///
28/// In helper functions that convert the response body to a value the [`HttpError`] is converted into this type. You
29/// can downcast to get the HTTP error in those cases.
30pub type Error = Box<dyn std::error::Error + Send + Sync>;
31
32use futures_lite::AsyncBufReadExt;
33pub use http::{
34 StatusCode, header,
35 method::{self, Method},
36 uri::{self, Uri},
37};
38use serde::{Deserialize, Serialize};
39use zng_var::{Var, const_var};
40
41use std::{borrow::Cow, time::Duration};
42use std::{fmt, mem};
43
44use crate::{channel::IpcBytes, http::ctx::REQUEST_DEFAULT, io::Metrics};
45
46use super::io::AsyncRead;
47
48use zng_txt::{ToTxt, Txt, formatx};
49use zng_unit::*;
50
51/// HTTP request.
52///
53/// Use [`send`] to send a request.
54#[derive(Debug, Clone, Serialize, Deserialize)]
55#[non_exhaustive]
56pub struct Request {
57 /// The URI.
58 #[serde(with = "http_serde::uri")]
59 pub uri: Uri,
60 /// The HTTP method.
61 #[serde(with = "http_serde::method")]
62 pub method: Method,
63
64 /// Header values.
65 ///
66 /// Is empty by default.
67 #[serde(with = "http_serde::header_map")]
68 pub headers: http::HeaderMap,
69
70 /// Maximum amount of time that a complete request/response cycle is allowed to
71 /// take before being aborted. This includes DNS resolution, connecting to the server,
72 /// writing the request, and reading the response.
73 ///
74 /// Note that this includes the response read operation, so if you get a response but don't
75 /// read-it within this timeout you will get a [`TimedOut`] IO error.
76 ///
77 /// By default no timeout is used, [`Duration::MAX`].
78 ///
79 /// [`TimedOut`]: https://doc.rust-lang.org/nightly/std/io/enum.ErrorKind.html#variant.TimedOut
80 pub timeout: Duration,
81
82 /// Maximum amount of time to await for establishing connections to a host.
83 ///
84 /// Is 90 seconds by default.
85 pub connect_timeout: Duration,
86
87 /// Maximum amount of time allowed when transfer speed is under the given speed in bytes per second.
88 ///
89 /// By default not timeout is used, `(Duration::MAX, 0)`.
90 pub low_speed_timeout: (Duration, ByteLength),
91
92 /// Maximum redirects to follow.
93 ///
94 /// When redirecting the `Referer` header is updated automatically.
95 ///
96 /// Is `20` by default.
97 pub redirect_limit: u16,
98
99 /// If should auto decompress received data.
100 ///
101 /// If enabled the "Accept-Encoding" will also be set automatically, if it was not set on the header.
102 ///
103 /// This is enabled by default.
104 #[cfg(feature = "http_compression")]
105 pub auto_decompress: bool,
106
107 /// Maximum upload speed in bytes per second.
108 ///
109 /// No maximum by default, [`ByteLength::MAX`].
110 pub max_upload_speed: ByteLength,
111
112 /// Maximum download speed in bytes per second.
113 ///
114 /// No maximum by default, [`ByteLength::MAX`].
115 pub max_download_speed: ByteLength,
116
117 /// If the `Content-Length` header must be present in the response.
118 ///
119 /// By default this is not required.
120 pub require_length: bool,
121
122 /// Set the maximum response content length allowed.
123 ///
124 /// If the `Content-Length` is present on the response and it exceeds this limit an error is
125 /// returned immediately, otherwise if [`require_length`] is not enabled an error will be returned
126 /// only when the downloaded body length exceeds the limit.
127 ///
128 /// By default no limit is set, [`ByteLength::MAX`].
129 ///
130 /// [`require_length`]: Request::require_length
131 pub max_length: ByteLength,
132
133 /// Response cache mode.
134 ///
135 /// Is [`CacheMode::Default`] by default.
136 pub cache: CacheMode,
137
138 /// If cookies should be send and stored.
139 ///
140 /// When enabled the [`http_cache`] is used to retrieve and store cookies.
141 ///
142 /// Is not enabled by default.
143 #[cfg(feature = "http_cookie")]
144 pub cookies: bool,
145
146 /// If transfer metrics should be measured.
147 ///
148 /// When enabled you can get the information using the [`Response::metrics`] method.
149 ///
150 /// This is enabled by default.
151 pub metrics: bool,
152
153 /// Request body content.
154 ///
155 /// Is empty by default.
156 pub body: IpcBytes,
157}
158impl Request {
159 /// Starts building a request.
160 ///
161 /// # Examples
162 ///
163 /// ```
164 /// use zng_task::http;
165 ///
166 /// # fn try_example() -> Result<(), http::Error> {
167 /// let request = http::Request::new(http::Method::PUT, "https://httpbin.org/put".try_into()?);
168 /// # Ok(()) }
169 /// ```
170 pub fn new(method: Method, uri: Uri) -> Self {
171 match REQUEST_DEFAULT.lock().clone() {
172 Some(mut r) => {
173 r.method = method;
174 r.uri = uri;
175 r
176 }
177 None => Self {
178 uri,
179 method,
180 require_length: false,
181 max_length: ByteLength::MAX,
182 headers: header::HeaderMap::new(),
183 timeout: Duration::MAX,
184 connect_timeout: 90.secs(),
185 low_speed_timeout: (Duration::MAX, 0.bytes()),
186 redirect_limit: 20,
187 #[cfg(feature = "http_compression")]
188 auto_decompress: true,
189 max_upload_speed: ByteLength::MAX,
190 max_download_speed: ByteLength::MAX,
191 cache: CacheMode::Default,
192 #[cfg(feature = "http_cookie")]
193 cookies: false,
194 metrics: true,
195 body: IpcBytes::default(),
196 },
197 }
198 }
199
200 /// Starts building a GET request.
201 ///
202 /// # Examples
203 ///
204 /// ```
205 /// use zng_task::http;
206 ///
207 /// # fn try_example() -> Result<(), http::Error> {
208 /// let get = http::Request::get("https://httpbin.org/get")?;
209 /// # Ok(()) }
210 /// ```
211 pub fn get<U: TryInto<Uri>>(uri: U) -> Result<Self, <U as TryInto<Uri>>::Error> {
212 Ok(Self::new(Method::GET, uri.try_into()?))
213 }
214
215 /// Starts building a PUT request.
216 ///
217 /// # Examples
218 ///
219 /// ```
220 /// use zng_task::http;
221 ///
222 /// # fn try_example() -> Result<(), http::Error> {
223 /// let put = http::Request::put("https://httpbin.org/put")?.header("accept", "application/json")?;
224 /// # Ok(()) }
225 /// ```
226 pub fn put<U: TryInto<Uri>>(uri: U) -> Result<Self, <U as TryInto<Uri>>::Error> {
227 Ok(Self::new(Method::PUT, uri.try_into()?))
228 }
229
230 /// Starts building a POST request.
231 ///
232 /// # Examples
233 ///
234 /// ```
235 /// use zng_task::http;
236 ///
237 /// # fn try_example() -> Result<(), http::Error> {
238 /// let post = http::Request::post("https://httpbin.org/post")?.header("accept", "application/json")?;
239 /// # Ok(()) }
240 /// ```
241 pub fn post<U: TryInto<Uri>>(uri: U) -> Result<Self, <U as TryInto<Uri>>::Error> {
242 Ok(Self::new(Method::POST, uri.try_into()?))
243 }
244
245 /// Starts building a DELETE request.
246 ///
247 /// # Examples
248 ///
249 /// ```
250 /// use zng_task::http;
251 ///
252 /// # fn try_example() -> Result<(), http::Error> {
253 /// let delete = http::Request::delete("https://httpbin.org/delete")?.header("accept", "application/json")?;
254 /// # Ok(()) }
255 /// ```
256 pub fn delete<U: TryInto<Uri>>(uri: U) -> Result<Self, <U as TryInto<Uri>>::Error> {
257 Ok(Self::new(Method::DELETE, uri.try_into()?))
258 }
259
260 /// Starts building a PATCH request.
261 ///
262 /// # Examples
263 ///
264 /// ```
265 /// use zng_task::http;
266 ///
267 /// # fn try_example() -> Result<(), http::Error> {
268 /// let patch = http::Request::patch("https://httpbin.org/patch")?.header("accept", "application/json")?;
269 /// # Ok(()) }
270 /// ```
271 pub fn patch<U: TryInto<Uri>>(uri: U) -> Result<Self, <U as TryInto<Uri>>::Error> {
272 Ok(Self::new(Method::PATCH, uri.try_into()?))
273 }
274
275 /// Starts building a HEAD request.
276 ///
277 /// # Examples
278 ///
279 /// ```
280 /// use zng_task::http;
281 ///
282 /// # fn try_example() -> Result<(), http::Error> {
283 /// let head = http::Request::head("https://httpbin.org")?;
284 /// # Ok(()) }
285 /// ```
286 pub fn head<U: TryInto<Uri>>(uri: U) -> Result<Self, <U as TryInto<Uri>>::Error> {
287 Ok(Self::new(Method::HEAD, uri.try_into()?))
288 }
289
290 /// Appends a header to [`headers`] to this request.
291 ///
292 /// [`headers`]: field@Request::headers
293 pub fn header<K, V>(mut self, name: K, value: V) -> Result<Self, Error>
294 where
295 K: TryInto<header::HeaderName>,
296 V: TryInto<header::HeaderValue>,
297 Error: From<<K as TryInto<header::HeaderName>>::Error>,
298 Error: From<<V as TryInto<header::HeaderValue>>::Error>,
299 {
300 self.headers.insert(name.try_into()?, value.try_into()?);
301 Ok(self)
302 }
303
304 /// Append Server-Sent Events (SSE) headers and disables cache.
305 ///
306 /// Set the `last_event_id` if this request is to reconnect a previous stream that tracks messages received.
307 ///
308 /// # Examples
309 ///
310 /// ```
311 /// use zng_task::http;
312 ///
313 /// # fn try_example() -> Result<(), http::Error> {
314 /// let sse = http::Request::get("https://httpbun.com/sse")?.sse("")?;
315 /// # Ok(()) }
316 /// ```
317 pub fn sse(mut self, last_event_id: &str) -> Result<Self, Error> {
318 self.cache = CacheMode::NoCache;
319 let r = self.header("accept", "text/event-stream")?;
320 if last_event_id.is_empty() {
321 Ok(r)
322 } else {
323 r.header("Last-Event-ID", last_event_id)
324 }
325 }
326
327 /// Set the [`timeout`].
328 ///
329 /// [`timeout`]: field@Request::timeout
330 pub fn timeout(mut self, timeout: Duration) -> Self {
331 self.timeout = timeout;
332 self
333 }
334
335 /// Set the [`connect_timeout`].
336 ///
337 /// [`connect_timeout`]: field@Request::connect_timeout
338 pub fn connect_timeout(mut self, timeout: Duration) -> Self {
339 self.connect_timeout = timeout;
340 self
341 }
342
343 /// Set the [`low_speed_timeout`].
344 ///
345 /// [`low_speed_timeout`]: field@Request::low_speed_timeout
346 pub fn low_speed_timeout(mut self, timeout: Duration, bytes_per_sec: ByteLength) -> Self {
347 self.low_speed_timeout = (timeout, bytes_per_sec);
348 self
349 }
350
351 /// Set the [`redirect_limit`].
352 ///
353 /// [`redirect_limit`]: field@Request::redirect_limit
354 pub fn redirect_limit(mut self, count: u16) -> Self {
355 self.redirect_limit = count;
356 self
357 }
358
359 /// Set the [`auto_decompress`].
360 ///
361 /// [`auto_decompress`]: field@Request::auto_decompress
362 #[cfg(feature = "http_compression")]
363 pub fn auto_decompress(mut self, enabled: bool) -> Self {
364 self.auto_decompress = enabled;
365 self
366 }
367
368 /// Set [`require_length`].
369 ///
370 /// [`require_length`]: field@Request::require_length
371 pub fn require_length(mut self, enabled: bool) -> Self {
372 self.require_length = enabled;
373 self
374 }
375
376 /// Set [`max_length`].
377 ///
378 /// [`max_length`]: field@Request::max_length
379 pub fn max_length(mut self, max: ByteLength) -> Self {
380 self.max_length = max;
381 self
382 }
383
384 /// Set the [`max_upload_speed`].
385 ///
386 /// [`max_upload_speed`]: field@Request::max_upload_speed
387 pub fn max_upload_speed(mut self, bytes_per_sec: ByteLength) -> Self {
388 self.max_upload_speed = bytes_per_sec;
389 self
390 }
391
392 /// Set the [`max_download_speed`].
393 ///
394 /// [`max_download_speed`]: field@Request::max_download_speed
395 pub fn max_download_speed(mut self, bytes_per_sec: ByteLength) -> Self {
396 self.max_download_speed = bytes_per_sec;
397 self
398 }
399
400 /// Set the [`cookies`].
401 ///
402 /// [`cookies`]: field@Request::cookies
403 #[cfg(feature = "http_cookie")]
404 pub fn cookies(mut self, enable: bool) -> Self {
405 self.cookies = enable;
406 self
407 }
408
409 /// Set the [`metrics`].
410 ///
411 /// [`metrics`]: field@Request::metrics
412 pub fn metrics(mut self, enabled: bool) -> Self {
413 self.metrics = enabled;
414 self
415 }
416
417 /// Set the [`body`].
418 ///
419 /// [`body`]: field@Request::body
420 pub fn body(mut self, body: IpcBytes) -> Self {
421 self.body = body;
422 self
423 }
424
425 /// Set the [`body`] to a plain text UTF-8 payload. Also sets the `Content-Type` header if it is not set.
426 ///
427 /// [`body`]: field@Request::body
428 pub fn body_text(mut self, body: &str) -> Result<Self, Error> {
429 if !self.headers.contains_key("Content-Type") {
430 self = self.header("Content-Type", "text/plain; charset=utf-8")?;
431 }
432 Ok(self.body(IpcBytes::from_slice_blocking(body.as_bytes())?))
433 }
434
435 /// Set the [`body`] to a JSON payload. Also sets the `Content-Type` header if it is not set.
436 ///
437 /// [`body`]: field@Request::body
438 pub fn body_json<T: Serialize>(mut self, body: &T) -> Result<Self, Error> {
439 if !self.headers.contains_key("Content-Type") {
440 self = self.header("Content-Type", "text/json; charset=utf-8")?;
441 }
442 let body = serde_json::to_vec(body)?;
443 Ok(self.body(IpcBytes::from_vec_blocking(body)?))
444 }
445}
446impl From<Request> for http::Request<IpcBytes> {
447 fn from(mut r: Request) -> Self {
448 let mut b = http::Request::builder().uri(mem::take(&mut r.uri)).method(r.method.clone());
449 if !r.headers.is_empty() {
450 *b.headers_mut().unwrap() = mem::take(&mut r.headers);
451 }
452 let body = mem::take(&mut r.body);
453 let b = b.extension(r);
454 b.body(body).unwrap()
455 }
456}
457impl From<http::Request<IpcBytes>> for Request {
458 fn from(value: http::Request<IpcBytes>) -> Self {
459 let (mut parts, body) = value.into_parts();
460 if let Some(mut r) = parts.extensions.remove::<Request>() {
461 r.method = parts.method;
462 r.uri = parts.uri;
463 r.headers = parts.headers;
464 r.body = body;
465 r
466 } else {
467 let mut r = Request::new(parts.method, parts.uri);
468 r.headers = parts.headers;
469 r.body = body;
470 r
471 }
472 }
473}
474
475/// HTTP response.
476pub struct Response {
477 status: StatusCode,
478 headers: header::HeaderMap,
479 effective_uri: Uri,
480 body: ResponseBody,
481 metrics: Var<Metrics>,
482}
483impl fmt::Debug for Response {
484 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
485 f.debug_struct("Response")
486 .field("status", &self.status)
487 .field("effective_uri", &self.effective_uri)
488 .field("header", &self.headers)
489 .field("metrics", &self.metrics.get())
490 .finish_non_exhaustive()
491 }
492}
493enum ResponseBody {
494 Done { bytes: IpcBytes },
495 Read { read: Box<dyn AsyncRead + Send> },
496}
497impl Response {
498 /// New with body download pending or ongoing.
499 pub fn from_read(
500 status: StatusCode,
501 header: header::HeaderMap,
502 effective_uri: Uri,
503 metrics: Var<Metrics>,
504 read: Box<dyn AsyncRead + Send>,
505 ) -> Self {
506 Self {
507 status,
508 headers: header,
509 effective_uri,
510 metrics,
511 body: ResponseBody::Read { read },
512 }
513 }
514
515 /// New with body already downloaded.
516 pub fn from_done(status: StatusCode, mut headers: header::HeaderMap, effective_uri: Uri, metrics: Metrics, body: IpcBytes) -> Self {
517 if !headers.contains_key(header::CONTENT_LENGTH) {
518 headers.insert(header::CONTENT_LENGTH, body.len().into());
519 }
520 Self {
521 status,
522 headers,
523 effective_uri,
524 metrics: const_var(metrics),
525 body: ResponseBody::Done { bytes: body },
526 }
527 }
528
529 /// New with status and message body.
530 pub fn from_msg(status: StatusCode, msg: impl ToTxt) -> Self {
531 Self::from_done(
532 status,
533 header::HeaderMap::new(),
534 Uri::from_static("/"),
535 Metrics::zero(),
536 IpcBytes::from_slice_blocking(msg.to_txt().as_bytes()).unwrap(),
537 )
538 }
539
540 /// Returns the [`StatusCode`].
541 pub fn status(&self) -> StatusCode {
542 self.status
543 }
544
545 /// Returns a reference to the associated header field map.
546 pub fn header(&self) -> &header::HeaderMap {
547 &self.headers
548 }
549
550 /// Get the effective URI of this response. This value differs from the
551 /// original URI provided when making the request if at least one redirect
552 /// was followed.
553 pub fn effective_uri(&self) -> &Uri {
554 &self.effective_uri
555 }
556
557 /// Get the body bytes length if it is downloaded or `Content-Length` value if it is present in the headers.
558 pub fn content_len(&self) -> Option<ByteLength> {
559 match &self.body {
560 ResponseBody::Done { bytes, .. } => Some((bytes.len() as u64).bytes()),
561 ResponseBody::Read { .. } => {
562 let len = self
563 .headers
564 .get(header::CONTENT_LENGTH)?
565 .to_str()
566 .ok()?
567 .parse::<u64>()
568 .ok()?
569 .bytes();
570 Some(len)
571 }
572 }
573 }
574
575 /// Receive the full body to memory.
576 pub async fn download(&mut self) -> Result<(), Error> {
577 if let ResponseBody::Done { .. } = &self.body {
578 return Ok(());
579 }
580
581 let downloader = match mem::replace(
582 &mut self.body,
583 ResponseBody::Done {
584 bytes: IpcBytes::default(),
585 },
586 ) {
587 ResponseBody::Read { read: downloader } => downloader,
588 ResponseBody::Done { .. } => unreachable!(),
589 };
590 let mut downloader = Box::into_pin(downloader);
591 let body = IpcBytes::from_read(downloader.as_mut()).await?;
592
593 self.body = ResponseBody::Done { bytes: body };
594
595 Ok(())
596 }
597
598 /// Download the full body memory cache and returns a reference to it.
599 pub async fn body(&mut self) -> Result<IpcBytes, Error> {
600 self.download().await?;
601 match &self.body {
602 ResponseBody::Done { bytes } => Ok(bytes.clone()),
603 ResponseBody::Read { .. } => unreachable!(),
604 }
605 }
606
607 /// Download the full body to memory and decode a clone to text.
608 ///
609 /// The text encoding is guessed from header `"content-type"` and *BOM sniffing*, otherwise
610 /// utf-8 is assumed and the replacement character (U+FFFD) is used for invalid chars.
611 pub async fn body_text(&mut self) -> Result<Txt, Error> {
612 let content_type = self
613 .headers
614 .get(header::CONTENT_TYPE)
615 .and_then(|value| value.to_str().ok())
616 .and_then(|value| value.parse::<mime::Mime>().ok());
617 let encoding_name = content_type
618 .as_ref()
619 .and_then(|mime| mime.get_param("charset").map(|charset| charset.as_str()))
620 .unwrap_or("utf-8");
621
622 let bytes = self.body().await?;
623
624 let encoding = encoding_rs::Encoding::for_label(encoding_name.as_bytes()).unwrap_or(encoding_rs::UTF_8);
625 let (text, _, _) = encoding.decode(&bytes);
626 Ok(Txt::from_str(&text))
627 }
628
629 /// Download the full body to memory and deserialize it from JSON.
630 pub async fn body_json<O>(&mut self) -> Result<O, Error>
631 where
632 O: serde::de::DeserializeOwned + std::marker::Unpin,
633 {
634 let bytes = self.body().await?;
635 let r = serde_json::from_slice(&bytes)?;
636 Ok(r)
637 }
638
639 /// Take the body stream.
640 ///
641 /// After this call the `body_*` methods will return empty.
642 ///
643 /// If the body was already downloaded reads that data from the start.
644 ///
645 /// Note that the [`metrics`] var will continue updating for the stream.
646 ///
647 /// [`metrics`]: Self::metrics
648 pub fn take_stream(&mut self) -> Box<dyn AsyncRead + Send> {
649 match mem::replace(
650 &mut self.body,
651 ResponseBody::Done {
652 bytes: IpcBytes::default(),
653 },
654 ) {
655 ResponseBody::Done { bytes } => Box::new(crate::io::Cursor::new(bytes)),
656 ResponseBody::Read { read } => read,
657 }
658 }
659
660 /// Take the body stream and decodes it to UTF-8.
661 ///
662 /// The text encoding is guessed from header `"content-type"` and *BOM sniffing*, otherwise
663 /// utf-8 is assumed and the replacement character (U+FFFD) is used for invalid chars.
664 pub fn stream_decode(&mut self) -> impl AsyncRead + Send + 'static {
665 let content_type = self
666 .headers
667 .get(header::CONTENT_TYPE)
668 .and_then(|value| value.to_str().ok())
669 .and_then(|value| value.parse::<mime::Mime>().ok());
670 let encoding_name = content_type
671 .as_ref()
672 .and_then(|mime| mime.get_param("charset").map(|charset| charset.as_str()))
673 .unwrap_or("utf-8");
674 let encoding = encoding_rs::Encoding::for_label(encoding_name.as_bytes()).unwrap_or(encoding_rs::UTF_8);
675
676 stream_decoder::StreamDecoder::new(self.take_stream(), encoding.new_decoder())
677 }
678
679 /// Take the body stream and decode Server-Sent Events (SSE).
680 ///
681 /// The `handler` is called for each [`SseArgs`] and once at the end with [`SseDisconnected`].
682 /// The `handler` is dropped after disconnect, the caller must make a new request to retry if desired.
683 ///
684 /// If the `handler` returns `false` it is dropped and disconnected.
685 pub fn on_sse<F>(&mut self, mut handler: impl FnMut(Result<SseArgs, SseDisconnected>) -> F + Send + 'static)
686 where
687 F: Future<Output = bool> + Send,
688 {
689 // no need to decode, sse is always utf-8
690 let mut stream = crate::io::BufReader::new(Box::into_pin(self.take_stream()));
691 let mut first = true;
692
693 crate::spawn(async move {
694 let mut buf = vec![];
695 let mut last_event_id = String::new();
696 let mut last_retry = String::new();
697 let mut error = None;
698 loop {
699 // read_line has a bug https://github.com/smol-rs/futures-lite/issues/143
700 // it also allocates a second buf for some reason, read_until is more efficient
701 let mut result = stream.read_until(b'\n', &mut buf).await;
702 if let &Ok(n) = &result
703 && n > 0
704 {
705 // validate the line
706 let start = buf.len() - n;
707 if let Err(e) = str::from_utf8(&buf[start..]) {
708 result = Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e));
709 }
710 }
711 match result {
712 Ok(n) => {
713 // SAFETY: we validate by line as they are received
714 let mut buf_str = unsafe { str::from_utf8_unchecked(&buf) };
715 if mem::take(&mut first) {
716 // strip utf-8 BOM
717 buf_str = buf_str.strip_prefix('\u{FEFF}').unwrap_or(buf_str);
718 }
719
720 if matches!(buf_str.lines().next_back(), Some(l) if l.is_empty()) {
721 // received empty line
722
723 let mut id = None;
724 let mut event = None;
725 let mut retry = None;
726 let mut data = None::<Cow<str>>;
727 for line in buf_str.lines() {
728 // spec says a line without : is a field with no value,
729 // so "data\n\n" is the same as "data:\n\n"
730 let (field, s) = line.split_once(':').unwrap_or((line, ""));
731 let s = s.strip_prefix(' ').unwrap_or(s);
732 match field {
733 "data" => {
734 if let Some(d) = &mut data {
735 let d = d.to_mut();
736 d.push('\n');
737 d.push_str(s);
738 } else {
739 data = Some(Cow::Borrowed(s));
740 }
741 }
742 "event" => event = Some(s),
743 "id" => {
744 if !s.contains('\0') {
745 id = Some(s)
746 }
747 }
748 "retry" if !s.is_empty() && s.chars().all(|c| c.is_ascii_digit()) => retry = Some(s),
749 _ => {}
750 }
751 }
752
753 if let Some(id) = id {
754 last_event_id.clear();
755 last_event_id.push_str(id);
756 }
757 if let Some(ms) = retry {
758 last_retry.clear();
759 last_retry.push_str(ms);
760 }
761 if let Some(data) = data {
762 let event = event.unwrap_or("");
763 let continue_ = handler(Ok(SseArgs { event, data: &data, id })).await;
764 if !continue_ {
765 // disconnect by user, skip `Disconnected` event
766 return;
767 }
768 }
769
770 buf.clear();
771 }
772
773 if n == 0 {
774 // disconnected by EOF, goto `Disconnected` event
775 if !buf.is_empty() {
776 // spec says last event is terminated by \n\n too
777 error = Some(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "EOF before last \\n\\n"))
778 }
779 break;
780 }
781 }
782 Err(e) => {
783 error = Some(e);
784 // disconnected by error, goto `Disconnected` event
785 break;
786 }
787 }
788 }
789 let mut retry_delay = Duration::ZERO;
790 if !last_retry.is_empty()
791 && let Ok(ms) = last_retry.parse()
792 {
793 retry_delay = Duration::from_millis(ms);
794 }
795 handler(Err(SseDisconnected {
796 last_event_id: &last_event_id,
797 retry_delay,
798 error,
799 }))
800 .await;
801 });
802 }
803
804 /// Metrics for the task transfer, if it was enabled in the request.
805 pub fn metrics(&self) -> Var<Metrics> {
806 self.metrics.read_only()
807 }
808
809 /// Convert failed status to an error.
810 ///
811 /// If the [`status`] is not success this will attempt to download the body as text, otherwise
812 /// the body is not accessed.
813 ///
814 /// [`status`]: Response::status
815 pub async fn error(&mut self) -> Result<(), HttpError> {
816 if self.status.is_success() {
817 Ok(())
818 } else {
819 let body = match self.body_text().await {
820 Ok(b) => b,
821 Err(e) => formatx!("could not receive error body, {e}"),
822 };
823 Err(HttpError { status: self.status, body })
824 }
825 }
826}
827
828/// Represents an HTTP response error.
829///
830/// See [`Response::error`] for more details.
831#[derive(Debug, Clone, PartialEq)]
832#[non_exhaustive]
833pub struct HttpError {
834 /// The error code.
835 pub status: StatusCode,
836 /// The error body.
837 pub body: Txt,
838}
839
840impl HttpError {
841 /// New error.
842 pub fn new(status: StatusCode, body: Txt) -> Self {
843 Self { status, body }
844 }
845}
846/// Alternate writes the `body`.
847impl fmt::Display for HttpError {
848 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
849 write!(f, "HTTP error {}", self.status.as_str())?;
850 if let Some(r) = self.status.canonical_reason() {
851 write!(f, " {r}")?;
852 }
853 if f.alternate() && self.body.is_empty() {
854 for line in self.body.lines() {
855 write!(f, "\n {line}")?;
856 }
857 }
858 Ok(())
859 }
860}
861impl std::error::Error for HttpError {}
862
863/// Send a GET request to the `uri`.
864///
865/// The [`http_client`] is used to send the request.
866pub async fn get<U>(uri: U) -> Result<Response, Error>
867where
868 U: TryInto<Uri>,
869 Error: From<<U as TryInto<Uri>>::Error>,
870{
871 send(Request::get(uri)?).await
872}
873
874/// Send a GET request to the `uri` and read the response as a string.
875///
876/// The [`http_client`] is used to send the request.
877pub async fn get_txt<U>(uri: U) -> Result<Txt, Error>
878where
879 U: TryInto<Uri>,
880 Error: From<<U as TryInto<Uri>>::Error>,
881{
882 let mut r = send(Request::get(uri)?).await?;
883 r.error().await?;
884 r.body_text().await
885}
886
887/// Send a GET request to the `uri` and read the response as raw bytes.
888///
889/// The [`http_client`] is used to send the request.
890pub async fn get_bytes<U>(uri: U) -> Result<IpcBytes, Error>
891where
892 U: TryInto<Uri>,
893 Error: From<<U as TryInto<Uri>>::Error>,
894{
895 let mut r = send(Request::get(uri)?).await?;
896 r.error().await?;
897 r.body().await
898}
899
900/// Send a GET request to the `uri` and de-serializes the response.
901///
902/// The [`http_client`] is used to send the request.
903pub async fn get_json<U, O>(uri: U) -> Result<O, Error>
904where
905 U: TryInto<Uri>,
906 Error: From<<U as TryInto<Uri>>::Error>,
907 O: serde::de::DeserializeOwned + std::marker::Unpin,
908{
909 let mut r = send(Request::get(uri)?).await?;
910 r.error().await?;
911 r.body_json().await
912}
913
914/// Send a HEAD request to the `uri`.
915///
916/// The [`http_client`] is used to send the request.
917pub async fn head<U>(uri: U) -> Result<Response, Error>
918where
919 U: TryInto<Uri>,
920 Error: From<<U as TryInto<Uri>>::Error>,
921{
922 send(Request::head(uri)?).await
923}
924
925/// Send a PUT request to the `uri` with a given request body.
926///
927/// The [`http_client`] is used to send the request.
928pub async fn put<U>(uri: U, body: IpcBytes) -> Result<Response, Error>
929where
930 U: TryInto<Uri>,
931 Error: From<<U as TryInto<Uri>>::Error>,
932{
933 send(Request::put(uri)?.body(body)).await
934}
935
936/// Send a POST request to the `uri` with a given request body.
937///
938/// The [`http_client`] is used to send the request.
939pub async fn post<U>(uri: U, body: IpcBytes) -> Result<Response, Error>
940where
941 U: TryInto<Uri>,
942 Error: From<<U as TryInto<Uri>>::Error>,
943{
944 send(Request::post(uri)?.body(body)).await
945}
946
947/// Send a DELETE request to the `uri`.
948///
949/// The [`http_client`] is used to send the request.
950pub async fn delete<U>(uri: U) -> Result<Response, Error>
951where
952 U: TryInto<Uri>,
953 Error: From<<U as TryInto<Uri>>::Error>,
954{
955 send(Request::delete(uri)?).await
956}
957
958/// Send a custom [`Request`].
959///
960/// The [`http_client`] is used to send the request.
961pub async fn send(request: Request) -> Result<Response, Error> {
962 let client = http_client();
963 if client.is_cache_manager() {
964 client.send(request).await
965 } else {
966 match request.cache {
967 CacheMode::NoCache => client.send(request).await,
968 CacheMode::Default => cache::send_cache(client, request).await,
969 CacheMode::Permanent => cache::send_cache_perm(client, request).await,
970 }
971 }
972}
973
974/// Represents an update in a Server-Sent Events (SSE) stream.
975///
976/// See [`Response::on_sse`] for more details.
977#[derive(Debug)]
978pub struct SseArgs<'a> {
979 /// Id.
980 pub id: Option<&'a str>,
981 /// Event name.
982 ///
983 /// Empty means *message*.
984 pub event: &'a str,
985 /// Data.
986 pub data: &'a str,
987}
988impl<'a> SseArgs<'a> {
989 /// Gets if is not a custom event.
990 ///
991 /// Handling `data` only when this is `true` is the equivalent of
992 /// subscribing to `EventSource.onmessage` in Javascript.
993 pub fn is_message(&self) -> bool {
994 self.event.is_empty()
995 }
996
997 /// Gets if is custom event.
998 pub fn is_event(&self, name: &str) -> bool {
999 self.event == name
1000 }
1001
1002 /// Try to deserialize `data` from JSON.
1003 pub fn json<O: serde::Deserialize<'a>>(&self) -> Result<O, serde_json::Error> {
1004 serde_json::from_str(self.data)
1005 }
1006}
1007
1008/// Represents the end of a Server-Sent Events (SSE) stream.
1009///
1010/// See [`Response::on_sse`] for more details.
1011#[derive(Debug)]
1012pub struct SseDisconnected<'a> {
1013 /// Last event ID, use with [`Request::sse`] to reconnect at the same position.
1014 pub last_event_id: &'a str,
1015 /// Wait this delay before requesting a new connection.
1016 pub retry_delay: Duration,
1017 /// Error.
1018 ///
1019 /// This is `None` on EOF.
1020 pub error: Option<std::io::Error>,
1021}