io_http/
client.rs

1//! # HTTP client surfaces
2//!
3//! Two traits and one implementation of them. [`HttpClient`] and
4//! [`HttpClientAsync`] carry the request surface: implement the two
5//! pumps and inherit the commands. [`HttpClientStd`] is the opinionated
6//! blocking implementation, wrapping a boxed `Read + Write + Send`
7//! stream.
8//!
9//! Each [`send`] / [`send_http10`] is self-contained, HTTP having no
10//! session context. With a TLS feature enabled,
11//! [`HttpClientStd::connect`] opens `http://` / `https://` URLs
12//! end-to-end via [`pimalaya_stream::stream::Stream`].
13//!
14//! [`send`]: HttpClient::send
15//! [`send_http10`]: HttpClient::send_http10
16
17use core::{future::Future, mem};
18
19use alloc::{boxed::Box, string::String, vec, vec::Vec};
20
21use std::io::{self, Read, Write};
22
23use thiserror::Error;
24use url::Url;
25
26use crate::{
27    coroutine::*,
28    rfc1945::send::*,
29    rfc9110::{
30        headers::HTTP_TRANSFER_ENCODING,
31        request::HttpRequest,
32        response::HttpResponse,
33        send::{HttpSendOutput, HttpSendYield},
34    },
35    rfc9112::{chunk_stream::*, read_headers::*, send::*},
36    sse::frame::*,
37};
38
39#[cfg(any(
40    feature = "rustls-aws",
41    feature = "rustls-ring",
42    feature = "native-tls"
43))]
44mod connect;
45
46const READ_BUFFER_SIZE: usize = 16 * 1024;
47
48/// Errors returned by the client surfaces.
49#[derive(Debug, Error)]
50pub enum HttpClientError {
51    /// The HTTP/1.0 send coroutine failed.
52    #[error(transparent)]
53    Http10Send(#[from] Http10SendError),
54    /// The HTTP/1.1 send coroutine failed.
55    #[error(transparent)]
56    Http11Send(#[from] Http11SendError),
57    /// The underlying stream failed to read or write.
58    #[error(transparent)]
59    Io(#[from] io::Error),
60    /// The TCP connection or the TLS negotiation failed.
61    #[cfg(any(
62        feature = "rustls-aws",
63        feature = "rustls-ring",
64        feature = "native-tls"
65    ))]
66    #[error(transparent)]
67    Tls(#[from] anyhow::Error),
68    /// The URL to connect to carries no host.
69    #[error("HTTP URL `{0}` has no host")]
70    UrlMissingHost(String),
71    /// The URL to connect to carries a scheme the client cannot open.
72    #[error("HTTP URL `{0}` has unsupported scheme `{1}` (expected `http` or `https`)")]
73    UrlUnsupportedScheme(String, String),
74    /// The server answered with a redirect the client never follows.
75    #[error("HTTP server redirected to `{url}` (status `{code}`)")]
76    UnexpectedRedirect {
77        /// The resolved redirect target.
78        url: Url,
79        /// The 3xx status code of the response.
80        code: u16,
81    },
82    /// The streaming response did not use chunked transfer coding.
83    #[error("HTTP streaming requires `Transfer-Encoding: chunked` (got status `{0}`)")]
84    StreamingNotChunked(u16),
85    /// The streaming chunked-body decoder failed.
86    #[error(transparent)]
87    ChunkStream(#[from] Http11ChunksReadStreamError),
88    /// The implementor's own transport failed.
89    ///
90    /// [`HttpClientStd`] reports I/O through [`Self::Io`]; this variant
91    /// exists for implementors whose failures are something else, such
92    /// as a JNI upcall or a runtime-specific socket error.
93    #[error(transparent)]
94    Transport(Box<dyn core::error::Error + Send + Sync>),
95}
96
97/// Blocking HTTP request surface: implement the two pumps and inherit
98/// the commands.
99///
100/// [`HttpClientStd`] implements it over a `Read + Write` stream; a
101/// caller whose transport is its own (a JNI upcall bridge, an in-memory
102/// test double) implements the same two methods and gets the rest.
103///
104/// There are two pumps rather than one because HTTP has two yield
105/// vocabularies. [`run`] takes the plain read/write coroutines, the ones
106/// every client wraps identically. [`run_send`] takes the request
107/// coroutines, which also yield [`HttpSendYield::WantsRedirect`], and
108/// that yield is a policy question: this crate's own client refuses a
109/// redirect, a browser-shaped one would follow it, and a consumer
110/// bounded by an allow-list would inspect it. Making it a required
111/// method puts the decision in the implementor's hands and keeps it out
112/// of the defaults.
113///
114/// The trait is not dyn-compatible, because both pumps are generic. The
115/// dynamism this crate needs lives one layer down, at the boxed stream
116/// [`HttpClientStd`] holds.
117///
118/// [`run`]: Self::run
119/// [`run_send`]: Self::run_send
120pub trait HttpClient {
121    /// Runs a standard-shape coroutine to completion, fulfilling its
122    /// read and write requests against the transport.
123    fn run<C, T, E>(&mut self, coroutine: C) -> Result<T, HttpClientError>
124    where
125        C: HttpCoroutine<Yield = HttpYield, Return = Result<T, E>>,
126        HttpClientError: From<E>;
127
128    /// Runs a request coroutine to completion, deciding what a redirect
129    /// means along the way.
130    fn run_send<C, E>(&mut self, coroutine: C) -> Result<HttpSendOutput, HttpClientError>
131    where
132        C: HttpCoroutine<Yield = HttpSendYield, Return = Result<HttpSendOutput, E>>,
133        HttpClientError: From<E>;
134
135    /// Sends one HTTP/1.1 request and reads its response.
136    fn send(&mut self, request: HttpRequest) -> Result<HttpSendOutput, HttpClientError> {
137        self.run_send(Http11Send::new(request))
138    }
139
140    /// HTTP/1.0 counterpart of [`send`](Self::send).
141    fn send_http10(&mut self, request: HttpRequest) -> Result<HttpSendOutput, HttpClientError> {
142        self.run_send(Http10Send::new(request))
143    }
144}
145
146/// Async HTTP request surface, the [`HttpClient`] twin for callers
147/// whose transport is a future.
148///
149/// Everything [`HttpClient`] documents applies here, plus the `Send`
150/// bounds. They are load-bearing rather than defensive: a plain `async
151/// fn` in a trait cannot promise that the future it returns is `Send`,
152/// so anything built from the default bodies would fail to compile
153/// under `tokio::spawn`, which is the first thing a worker-spawning
154/// consumer reaches for. Declaring the return type explicitly as `impl
155/// Future<..> + Send`, with `Send` as a supertrait so `&mut Self`
156/// carries through, keeps the defaults spawnable.
157///
158/// [`HttpClient`] deliberately carries no such bound. A blocking call
159/// returns a value, so there is no future whose auto-traits need
160/// pinning down, and requiring `Send` there would exclude a perfectly
161/// good client built on a thread-affine handle.
162pub trait HttpClientAsync: Send {
163    /// Runs a standard-shape coroutine to completion, fulfilling its
164    /// read and write requests against the transport.
165    fn run<C, T, E>(
166        &mut self,
167        coroutine: C,
168    ) -> impl Future<Output = Result<T, HttpClientError>> + Send
169    where
170        C: HttpCoroutine<Yield = HttpYield, Return = Result<T, E>> + Send,
171        T: Send,
172        E: Send,
173        HttpClientError: From<E>;
174
175    /// Runs a request coroutine to completion, deciding what a redirect
176    /// means along the way.
177    fn run_send<C, E>(
178        &mut self,
179        coroutine: C,
180    ) -> impl Future<Output = Result<HttpSendOutput, HttpClientError>> + Send
181    where
182        C: HttpCoroutine<Yield = HttpSendYield, Return = Result<HttpSendOutput, E>> + Send,
183        E: Send,
184        HttpClientError: From<E>;
185
186    /// Sends one HTTP/1.1 request and reads its response.
187    fn send(
188        &mut self,
189        request: HttpRequest,
190    ) -> impl Future<Output = Result<HttpSendOutput, HttpClientError>> + Send {
191        self.run_send(Http11Send::new(request))
192    }
193
194    /// HTTP/1.0 counterpart of [`send`](Self::send).
195    fn send_http10(
196        &mut self,
197        request: HttpRequest,
198    ) -> impl Future<Output = Result<HttpSendOutput, HttpClientError>> + Send {
199        self.run_send(Http10Send::new(request))
200    }
201}
202
203/// Std-blocking HTTP client wrapping a boxed `Read + Write + Send` stream.
204pub struct HttpClientStd {
205    stream: Box<dyn HttpStream>,
206}
207
208impl HttpClientStd {
209    /// Wraps a pre-connected stream; caller handles TCP and TLS.
210    pub fn new<S: Read + Write + Send + 'static>(stream: S) -> Self {
211        Self {
212            stream: Box::new(stream),
213        }
214    }
215
216    /// Default ALPN identifier for HTTPS connections: `http/1.1`
217    /// ([RFC 7301] + IANA registry).
218    ///
219    /// [RFC 7301]: https://www.rfc-editor.org/rfc/rfc7301
220    pub fn default_alpn() -> Vec<String> {
221        vec![String::from("http/1.1")]
222    }
223
224    /// Replaces the underlying stream (e.g. after `Connection: close` or
225    /// a cross-authority redirect).
226    pub fn set_stream<S: Read + Write + Send + 'static>(&mut self, stream: S) {
227        self.stream = Box::new(stream);
228    }
229}
230
231impl HttpClient for HttpClientStd {
232    fn run<C, T, E>(&mut self, mut coroutine: C) -> Result<T, HttpClientError>
233    where
234        C: HttpCoroutine<Yield = HttpYield, Return = Result<T, E>>,
235        HttpClientError: From<E>,
236    {
237        let mut buf = [0u8; READ_BUFFER_SIZE];
238        let mut arg: Option<&[u8]> = None;
239
240        loop {
241            match coroutine.resume(arg.take()) {
242                HttpCoroutineState::Complete(Ok(out)) => return Ok(out),
243                HttpCoroutineState::Complete(Err(err)) => return Err(err.into()),
244                HttpCoroutineState::Yielded(HttpYield::WantsRead) => {
245                    let n = self.stream.read(&mut buf)?;
246                    arg = Some(&buf[..n]);
247                }
248                HttpCoroutineState::Yielded(HttpYield::WantsWrite(bytes)) => {
249                    self.stream.write_all(&bytes)?;
250                    arg = None;
251                }
252            }
253        }
254    }
255
256    /// Refuses a redirect with [`HttpClientError::UnexpectedRedirect`],
257    /// this client following none: a request carrying credentials must
258    /// not replay them against whatever host a 3xx names, and the
259    /// caller is the only party that knows whether the new target is
260    /// one it meant to talk to.
261    fn run_send<C, E>(&mut self, mut coroutine: C) -> Result<HttpSendOutput, HttpClientError>
262    where
263        C: HttpCoroutine<Yield = HttpSendYield, Return = Result<HttpSendOutput, E>>,
264        HttpClientError: From<E>,
265    {
266        let mut buf = [0u8; READ_BUFFER_SIZE];
267        let mut arg: Option<&[u8]> = None;
268
269        loop {
270            match coroutine.resume(arg.take()) {
271                HttpCoroutineState::Complete(Ok(out)) => return Ok(out),
272                HttpCoroutineState::Complete(Err(err)) => return Err(err.into()),
273                HttpCoroutineState::Yielded(HttpSendYield::WantsRead) => {
274                    let n = self.stream.read(&mut buf)?;
275                    arg = Some(&buf[..n]);
276                }
277                HttpCoroutineState::Yielded(HttpSendYield::WantsWrite(bytes)) => {
278                    self.stream.write_all(&bytes)?;
279                    arg = None;
280                }
281                HttpCoroutineState::Yielded(HttpSendYield::WantsRedirect {
282                    url, response, ..
283                }) => {
284                    return Err(HttpClientError::UnexpectedRedirect {
285                        url,
286                        code: *response.status,
287                    });
288                }
289            }
290        }
291    }
292}
293
294impl HttpClientStd {
295    /// Opens an HTTP/1.1 SSE stream; requires `Transfer-Encoding: chunked`.
296    /// Consumes `self` because the connection is dedicated to the stream.
297    pub fn send_streaming(self, request: HttpRequest) -> Result<SseStream, HttpClientError> {
298        let HttpClientStd { mut stream } = self;
299
300        let req_bytes = request.to_http_11_vec();
301        stream.write_all(&req_bytes)?;
302
303        let mut read_headers = Http11HeadersRead::default();
304        let mut buf = [0u8; READ_BUFFER_SIZE];
305        let mut arg: Option<&[u8]> = None;
306
307        let out = loop {
308            match read_headers.resume(arg.take()) {
309                HttpCoroutineState::Complete(Ok(out)) => break out,
310                HttpCoroutineState::Complete(Err(err)) => {
311                    return Err(Http11SendError::from(err).into());
312                }
313                HttpCoroutineState::Yielded(HttpYield::WantsRead) => {
314                    let n = stream.read(&mut buf)?;
315                    if n == 0 {
316                        return Err(Http11SendError::Eof.into());
317                    }
318                    arg = Some(&buf[..n]);
319                }
320                HttpCoroutineState::Yielded(HttpYield::WantsWrite(_)) => {
321                    unreachable!("Http11HeadersRead never writes");
322                }
323            }
324        };
325
326        let chunked = out
327            .response
328            .header(HTTP_TRANSFER_ENCODING)
329            .is_some_and(|enc| enc.eq_ignore_ascii_case("chunked"));
330
331        if !chunked {
332            return Err(HttpClientError::StreamingNotChunked(*out.response.status));
333        }
334
335        Ok(SseStream {
336            stream,
337            chunk_stream: Http11ChunksReadStream::default(),
338            sse_parser: SseFrameParser::default(),
339            pending: None,
340            preread: out.remaining,
341            response: out.response,
342            keep_alive: out.keep_alive,
343            done: false,
344        })
345    }
346}
347
348/// Long-lived HTTP/1.1 Server-Sent Events stream; each
349/// [`SseStream::next_frame`] / [`Iterator::next`] blocks until the next
350/// event arrives or the connection closes.
351pub struct SseStream {
352    stream: Box<dyn HttpStream>,
353    chunk_stream: Http11ChunksReadStream,
354    sse_parser: SseFrameParser,
355    pending: Option<Vec<u8>>,
356    preread: Vec<u8>,
357    response: HttpResponse,
358    keep_alive: bool,
359    done: bool,
360}
361
362impl SseStream {
363    /// Parsed response headers (body is the streaming channel itself).
364    pub fn response(&self) -> &HttpResponse {
365        &self.response
366    }
367
368    /// Whether the server signalled the connection can be reused.
369    pub fn keep_alive(&self) -> bool {
370        self.keep_alive
371    }
372
373    /// Last-event-id seen so far; supply via `Last-Event-ID` on reconnect.
374    pub fn last_event_id(&self) -> Option<&str> {
375        self.sse_parser.last_event_id()
376    }
377
378    /// Drives chunked + SSE decoding until the next event; [`None`] on
379    /// connection close or zero-length chunk terminator.
380    pub fn next_frame(&mut self) -> Result<Option<SseFrame>, HttpClientError> {
381        if self.done {
382            return Ok(None);
383        }
384
385        loop {
386            let arg = self.pending.take();
387            match self.sse_parser.resume(arg.as_deref()) {
388                HttpCoroutineState::Yielded(SseFrameParserYield::Frame(frame)) => {
389                    return Ok(Some(frame));
390                }
391                HttpCoroutineState::Yielded(SseFrameParserYield::WantsBytes) => {
392                    match self.pull_chunk()? {
393                        Some(body) => self.pending = Some(body),
394                        None => {
395                            self.done = true;
396                            return Ok(None);
397                        }
398                    }
399                }
400                HttpCoroutineState::Complete(never) => match never {},
401            }
402        }
403    }
404
405    /// Closes the underlying connection (equivalent to dropping `self`).
406    pub fn close(self) {
407        drop(self);
408    }
409
410    fn pull_chunk(&mut self) -> Result<Option<Vec<u8>>, HttpClientError> {
411        let mut tmp = [0u8; READ_BUFFER_SIZE];
412        let preread = mem::take(&mut self.preread);
413        let mut arg: Option<&[u8]> = if preread.is_empty() {
414            None
415        } else {
416            Some(&preread)
417        };
418
419        loop {
420            match self.chunk_stream.resume(arg.take()) {
421                HttpCoroutineState::Yielded(Http11ChunksReadStreamYield::Frame { body }) => {
422                    return Ok(Some(body));
423                }
424                HttpCoroutineState::Complete(Ok(_remaining)) => return Ok(None),
425                HttpCoroutineState::Yielded(Http11ChunksReadStreamYield::WantsRead) => {
426                    let n = self.stream.read(&mut tmp)?;
427                    if n == 0 {
428                        return Ok(None);
429                    }
430                    arg = Some(&tmp[..n]);
431                }
432                HttpCoroutineState::Complete(Err(err)) => return Err(err.into()),
433            }
434        }
435    }
436}
437
438impl Iterator for SseStream {
439    type Item = Result<SseFrame, HttpClientError>;
440
441    fn next(&mut self) -> Option<Self::Item> {
442        match self.next_frame() {
443            Ok(Some(frame)) => Some(Ok(frame)),
444            Ok(None) => None,
445            Err(err) => Some(Err(err)),
446        }
447    }
448}
449
450/// Marker for everything the client can run against; the `Send`
451/// supertrait propagates through the `Box<dyn HttpStream>` erasure so
452/// [`HttpClientStd`] stays `Send`.
453trait HttpStream: Read + Write + Send {}
454impl<T: Read + Write + Send + ?Sized> HttpStream for T {}
io_http/client.rs

io_http/
client.rs
