http_net/client.rs
1//! The client side of an HTTP/1.1 connection, without I/O.
2
3use alloc::vec::Vec;
4
5use crate::{
6 body::{Decoder, Encoder, Step, push_framing},
7 buffer::Buffer,
8 error::HttpError,
9 head::{
10 Asked, HeadKind, HeadScanner, Headers, ResponseHead, ResponseInfo, Scan, interpret_response,
11 },
12 limits::Limits,
13 message::{OutBody, Request, check_trailers, push_last_chunk_with_trailers},
14 method::Method,
15 server::Progress,
16};
17
18/// The error for sending a body the server declined (a final status in
19/// answer to `Expect: 100-continue`).
20const REFUSED: HttpError = HttpError::InvalidState(
21 "the server answered with a final status instead of 100 Continue; \
22 the body must not be sent",
23);
24
25/// What a [`ClientConn`] has for the caller after [`poll`](ClientConn::poll).
26///
27/// # Examples
28///
29/// ```
30/// use http_net::{ClientConn, ClientEvent, Request};
31///
32/// let mut conn = ClientConn::new();
33/// let mut out = Vec::new();
34/// conn.send_request(&Request::get("/").header("host", "example.com"), &mut out).unwrap();
35/// let _ = conn.feed(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n3\r\nabc\r\n0\r\n\r\n");
36/// let mut body = Vec::new();
37/// loop {
38/// match conn.poll().unwrap() {
39/// ClientEvent::Response(resp) => assert_eq!(resp.status(), 200),
40/// ClientEvent::Body(chunk) => body.extend_from_slice(chunk),
41/// ClientEvent::ResponseEnd => break,
42/// other => panic!("{other:?}"),
43/// }
44/// }
45/// assert_eq!(body, b"abc");
46/// ```
47#[non_exhaustive]
48#[derive(Debug)]
49pub enum ClientEvent<'a> {
50 /// The final response head arrived. Its body follows as
51 /// [`Body`](ClientEvent::Body) events, then
52 /// [`ResponseEnd`](ClientEvent::ResponseEnd). Interim (1xx) responses
53 /// before it are skipped unless
54 /// [`set_report_informational`](ClientConn::set_report_informational)
55 /// asked for them; either way they count against
56 /// [`Limits::max_interim`](crate::Limits::max_interim) and, together,
57 /// [`Limits::max_header_bytes`](crate::Limits::max_header_bytes).
58 Response(ResponseHead<'a>),
59 /// An interim response other than `101` (for example `103 Early Hints`,
60 /// or `102 Processing`), only with
61 /// [`set_report_informational(true)`](ClientConn::set_report_informational).
62 /// The final response follows.
63 Informational(ResponseHead<'a>),
64 /// The server answered `100 Continue` to a request built with
65 /// [`Request::expect_continue`](crate::Request::expect_continue): send
66 /// the body now.
67 Continue,
68 /// A run of body bytes, after chunked decoding.
69 Body(&'a [u8]),
70 /// The response's trailer fields, just before
71 /// [`ResponseEnd`](ClientEvent::ResponseEnd): only with
72 /// [`set_report_trailers(true)`](ClientConn::set_report_trailers) and a
73 /// non-empty trailer section.
74 Trailers(Headers<'a>),
75 /// The response is complete.
76 ResponseEnd,
77 /// More input is needed (or, for a body that runs until close,
78 /// [`feed_eof`](ClientConn::feed_eof) once the server closes).
79 NeedInput,
80 /// No request is outstanding; send the next one.
81 Idle,
82 /// The server switched protocols (101 to an upgrade request, or 2xx to
83 /// `CONNECT`); the response head was reported just before. Bytes of the
84 /// new protocol that already arrived are in
85 /// [`remaining_input`](ClientConn::remaining_input).
86 Upgraded,
87 /// The connection cannot carry another request: close the socket.
88 Closed,
89}
90
91#[derive(Clone, Copy, Debug, PartialEq, Eq)]
92enum State {
93 Idle,
94 Head,
95 /// An interim head was reported and is still pinned; the next poll
96 /// drops it and waits for the next head.
97 Interim,
98 Body,
99 Complete,
100 Upgraded,
101 Closed,
102}
103
104#[derive(Clone, Copy, Debug, PartialEq, Eq)]
105enum ReqWrite {
106 Streaming(Encoder),
107 Sent,
108}
109
110/// The client side of one HTTP/1.1 connection: a sans-IO state machine.
111///
112/// One request is outstanding at a time. [`send_request`](Self::send_request)
113/// serializes a [`Request`]; [`feed`](Self::feed) and [`poll`](Self::poll)
114/// parse the response. When the response allows it, the connection returns
115/// to [`ClientEvent::Idle`] and can send the next request (keep-alive).
116///
117/// The response is framed with the request in mind: a `HEAD` request's
118/// response has no body whatever its headers say, a `CONNECT` 2xx becomes a
119/// tunnel, and a response with neither `Content-Length` nor
120/// `Transfer-Encoding` runs until the server closes.
121///
122/// # Examples
123///
124/// ```
125/// use http_net::{ClientConn, ClientEvent, Request};
126///
127/// let mut conn = ClientConn::new();
128/// let mut out = Vec::new();
129/// conn.send_request(&Request::get("/a").header("host", "h"), &mut out).unwrap();
130/// assert_eq!(out, b"GET /a HTTP/1.1\r\nhost: h\r\n\r\n");
131/// let _ = conn.feed(b"HTTP/1.1 204 No Content\r\n\r\n");
132/// assert!(matches!(conn.poll().unwrap(), ClientEvent::Response(_)));
133/// assert!(matches!(conn.poll().unwrap(), ClientEvent::ResponseEnd));
134/// assert!(matches!(conn.poll().unwrap(), ClientEvent::Idle));
135/// ```
136pub struct ClientConn {
137 buf: Buffer,
138 limits: Limits,
139 scanner: HeadScanner,
140 trailers: HeadScanner,
141 decoder: Decoder,
142 info: ResponseInfo,
143 asked: Asked,
144 upgrade_asked: bool,
145 state: State,
146 req: ReqWrite,
147 keep_alive: bool,
148 /// Bytes of interim (1xx) heads seen for the current request. They
149 /// count against the header budget so that a server cannot keep the
150 /// client parsing forever without answering.
151 interim_bytes: usize,
152 /// Interim heads seen for the current request, against `max_interim`.
153 interim_count: u16,
154 report_interim: bool,
155 report_trailers: bool,
156 /// The request was sent with `Expect: 100-continue` and neither a `100`,
157 /// a final response, nor body bytes have happened since.
158 waiting_continue: bool,
159 /// A final response arrived while waiting for `100`: the promised body
160 /// must not be sent, and the connection closes after the response.
161 body_refused: bool,
162 /// The request body was declared chunked (trailers allowed).
163 req_chunked: bool,
164}
165
166impl Default for ClientConn {
167 fn default() -> Self {
168 Self::new()
169 }
170}
171
172impl core::fmt::Debug for ClientConn {
173 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
174 f.debug_struct("ClientConn")
175 .field("state", &self.state)
176 .field("request", &self.req)
177 .field("buffered", &self.buf.unread().len())
178 .field("limits", &self.limits)
179 .finish()
180 }
181}
182
183impl ClientConn {
184 /// A connection with the default [`Limits`].
185 ///
186 /// # Examples
187 ///
188 /// ```
189 /// let mut conn = http_net::ClientConn::new();
190 /// assert!(matches!(conn.poll(), Ok(http_net::ClientEvent::Idle)));
191 /// ```
192 #[must_use]
193 pub fn new() -> Self {
194 Self::with_limits(Limits::new())
195 }
196
197 /// A connection enforcing `limits` on responses.
198 ///
199 /// # Examples
200 ///
201 /// ```
202 /// let conn = http_net::ClientConn::with_limits(http_net::Limits::new().with_max_body(1 << 24));
203 /// assert_eq!(conn.limits().max_body(), 1 << 24);
204 /// ```
205 #[must_use]
206 pub fn with_limits(limits: Limits) -> Self {
207 Self {
208 buf: Buffer::new(limits.buffer_capacity()),
209 limits,
210 scanner: HeadScanner::new(),
211 trailers: HeadScanner::new(),
212 decoder: Decoder::new(),
213 info: ResponseInfo::EMPTY,
214 asked: Asked {
215 head: false,
216 connect: false,
217 },
218 upgrade_asked: false,
219 state: State::Idle,
220 req: ReqWrite::Sent,
221 keep_alive: true,
222 interim_bytes: 0,
223 interim_count: 0,
224 report_interim: false,
225 report_trailers: false,
226 waiting_continue: false,
227 body_refused: false,
228 req_chunked: false,
229 }
230 }
231
232 /// The limits this connection enforces.
233 ///
234 /// # Examples
235 ///
236 /// ```
237 /// assert_eq!(http_net::ClientConn::new().limits().max_headers(), 100);
238 /// ```
239 #[must_use]
240 pub fn limits(&self) -> &Limits {
241 &self.limits
242 }
243
244 /// Copies as much of `bytes` into the input buffer as fits; returns how
245 /// many were taken (see [`ServerConn::feed`](crate::ServerConn::feed)).
246 ///
247 /// # Examples
248 ///
249 /// ```
250 /// let mut conn = http_net::ClientConn::new();
251 /// assert_eq!(conn.feed(b"HTTP/1.1"), 8);
252 /// ```
253 #[must_use = "bytes beyond the returned count were not taken"]
254 pub fn feed(&mut self, bytes: &[u8]) -> usize {
255 self.buf.feed(bytes)
256 }
257
258 /// Spare space at the end of the input buffer; follow with
259 /// [`commit`](Self::commit).
260 ///
261 /// # Examples
262 ///
263 /// ```
264 /// let mut conn = http_net::ClientConn::new();
265 /// assert!(!conn.read_buf().is_empty());
266 /// ```
267 pub fn read_buf(&mut self) -> &mut [u8] {
268 self.buf.spare()
269 }
270
271 /// Records that `n` bytes were written into [`read_buf`](Self::read_buf).
272 ///
273 /// # Examples
274 ///
275 /// ```
276 /// let mut conn = http_net::ClientConn::new();
277 /// conn.read_buf()[..4].copy_from_slice(b"HTTP");
278 /// conn.commit(4);
279 /// ```
280 pub fn commit(&mut self, n: usize) {
281 self.buf.commit(n);
282 }
283
284 /// Records that the server closed its sending side: ends a body that
285 /// runs until close, or reports [`HttpError::UnexpectedEof`] mid-message.
286 ///
287 /// # Examples
288 ///
289 /// ```
290 /// use http_net::{ClientConn, ClientEvent, Request};
291 ///
292 /// let mut conn = ClientConn::new();
293 /// conn.send_request(&Request::get("/").header("host", "a"), &mut Vec::new()).unwrap();
294 /// let _ = conn.feed(b"HTTP/1.0 200 OK\r\n\r\nall of it");
295 /// assert!(matches!(conn.poll().unwrap(), ClientEvent::Response(_)));
296 /// assert!(matches!(conn.poll().unwrap(), ClientEvent::Body(b"all of it")));
297 /// assert!(matches!(conn.poll().unwrap(), ClientEvent::NeedInput));
298 /// conn.feed_eof();
299 /// assert!(matches!(conn.poll().unwrap(), ClientEvent::ResponseEnd));
300 /// assert!(matches!(conn.poll().unwrap(), ClientEvent::Closed));
301 /// ```
302 pub fn feed_eof(&mut self) {
303 self.buf.set_eof();
304 }
305
306 /// Writes `request` into `out` and starts waiting for its response.
307 ///
308 /// A full body is sent with `content-length` (omitted for an empty body
309 /// on methods that do not expect one); a streamed body continues with
310 /// [`send_body`](Self::send_body) and [`end_body`](Self::end_body).
311 ///
312 /// A request built with [`Request::expect_continue`] is sent head first:
313 /// [`is_waiting_for_continue`](Self::is_waiting_for_continue) is true
314 /// until the server answers `100 Continue` (reported as
315 /// [`ClientEvent::Continue`]: then send the body), a final status (then
316 /// the body must not be sent and the connection closes after the
317 /// response), or the caller sends the body anyway (RFC 9110 section
318 /// 10.1.1 allows it after a wait of its choosing).
319 ///
320 /// # Errors
321 ///
322 /// - the first error recorded by the [`Request`] builder;
323 /// - [`HttpError::MissingHost`] when the request has no `host` header;
324 /// - [`HttpError::InvalidState`] when a request is already outstanding,
325 /// the connection is closed, or an `expect_continue` request has no
326 /// streamed body (`content_length` or `chunked`);
327 /// - [`HttpError::UnexpectedData`] when response bytes nobody asked for
328 /// are buffered.
329 ///
330 /// # Examples
331 ///
332 /// ```
333 /// use http_net::{ClientConn, HttpError, Request};
334 ///
335 /// let mut conn = ClientConn::new();
336 /// let mut out = Vec::new();
337 /// assert_eq!(conn.send_request(&Request::get("/"), &mut out), Err(HttpError::MissingHost));
338 /// conn.send_request(&Request::post("/f").header("host", "a").body(&b"xyz"[..]), &mut out).unwrap();
339 /// assert_eq!(out, b"POST /f HTTP/1.1\r\nhost: a\r\ncontent-length: 3\r\n\r\nxyz");
340 /// ```
341 pub fn send_request(&mut self, request: &Request, out: &mut Vec<u8>) -> Result<(), HttpError> {
342 if let Some(e) = request.error() {
343 return Err(e);
344 }
345 match self.state {
346 State::Idle => {}
347 State::Closed | State::Upgraded => {
348 return Err(HttpError::InvalidState(
349 "the connection is closed or upgraded",
350 ));
351 }
352 _ => return Err(HttpError::InvalidState("a request is already outstanding")),
353 }
354 if !self.buf.unread().is_empty() {
355 return Err(HttpError::UnexpectedData);
356 }
357 if !request.has_host {
358 return Err(HttpError::MissingHost);
359 }
360 if request.expect_continue && matches!(request.body, OutBody::Full(_)) {
361 // The point of the expectation is to hold the body back until
362 // the server agrees; a full body would be written at once.
363 return Err(HttpError::InvalidState(
364 "expect_continue needs a streamed body (content_length or chunked), \
365 sent after ClientEvent::Continue",
366 ));
367 }
368 let method = request.method_kind;
369 let expects_body = matches!(method, Method::Post | Method::Put | Method::Patch);
370 let (encoder, declared) = match &request.body {
371 OutBody::Full(b) if b.is_empty() && !expects_body => (Encoder::Discard, None),
372 OutBody::Full(b) => (
373 Encoder::Length {
374 remaining: b.len() as u64,
375 },
376 Some(b.len() as u64),
377 ),
378 OutBody::Length(n) => (Encoder::Length { remaining: *n }, Some(*n)),
379 OutBody::Chunked => (Encoder::Chunked, None),
380 };
381 out.extend_from_slice(request.method().as_bytes());
382 out.push(b' ');
383 out.extend_from_slice(request.target().as_bytes());
384 out.extend_from_slice(b" HTTP/1.1\r\n");
385 out.extend_from_slice(request.fields());
386 push_framing(out, encoder, declared);
387 if request.close {
388 out.extend_from_slice(b"connection: close\r\n");
389 }
390 out.extend_from_slice(b"\r\n");
391 let mut encoder = encoder;
392 match &request.body {
393 OutBody::Full(b) => {
394 encoder.write(b, out)?;
395 self.req = ReqWrite::Sent;
396 }
397 _ => self.req = ReqWrite::Streaming(encoder),
398 }
399 self.asked = Asked {
400 head: method == Method::Head,
401 connect: method == Method::Connect,
402 };
403 self.upgrade_asked = request.upgrade;
404 self.keep_alive = !request.close;
405 self.req_chunked = matches!(request.body, OutBody::Chunked);
406 self.waiting_continue = request.expect_continue;
407 self.body_refused = false;
408 self.buf.unpin();
409 self.buf.pin_at_pos();
410 self.scanner.reset();
411 self.interim_bytes = 0;
412 self.interim_count = 0;
413 self.state = State::Head;
414 Ok(())
415 }
416
417 /// Writes body bytes of a streamed request.
418 ///
419 /// Sending body bytes while waiting for `100 Continue` ends the wait.
420 ///
421 /// # Errors
422 ///
423 /// [`HttpError::BodyLengthMismatch`] past the declared length;
424 /// [`HttpError::InvalidState`] when no streamed request is in progress,
425 /// or when the server answered an `expect_continue` request with a final
426 /// status (the body must not be sent then).
427 ///
428 /// # Examples
429 ///
430 /// ```
431 /// use http_net::{ClientConn, Request};
432 ///
433 /// let mut conn = ClientConn::new();
434 /// let mut out = Vec::new();
435 /// conn.send_request(&Request::post("/up").header("host", "a").chunked(), &mut out).unwrap();
436 /// conn.send_body(b"part", &mut out).unwrap();
437 /// conn.end_body(&mut out).unwrap();
438 /// assert!(out.ends_with(b"4\r\npart\r\n0\r\n\r\n"));
439 /// ```
440 pub fn send_body(&mut self, data: &[u8], out: &mut Vec<u8>) -> Result<(), HttpError> {
441 if self.body_refused {
442 return Err(REFUSED);
443 }
444 match &mut self.req {
445 ReqWrite::Streaming(encoder) => {
446 encoder.write(data, out)?;
447 self.waiting_continue = false;
448 Ok(())
449 }
450 ReqWrite::Sent => Err(HttpError::InvalidState("no streamed request in progress")),
451 }
452 }
453
454 /// Ends a streamed request body.
455 ///
456 /// # Errors
457 ///
458 /// [`HttpError::BodyLengthMismatch`] when fewer bytes than declared were
459 /// sent; [`HttpError::InvalidState`] when no streamed request is in
460 /// progress.
461 ///
462 /// # Examples
463 ///
464 /// ```
465 /// use http_net::{ClientConn, Request};
466 ///
467 /// let mut conn = ClientConn::new();
468 /// let mut out = Vec::new();
469 /// conn.send_request(&Request::post("/").header("host", "a").content_length(2), &mut out).unwrap();
470 /// conn.send_body(b"ok", &mut out).unwrap();
471 /// conn.end_body(&mut out).unwrap();
472 /// assert!(out.ends_with(b"\r\n\r\nok"));
473 /// ```
474 pub fn end_body(&mut self, out: &mut Vec<u8>) -> Result<(), HttpError> {
475 self.end_body_with_trailers(&[], out)
476 }
477
478 /// Ends a streamed chunked request body with a trailer section (see
479 /// [`ServerConn::end_body_with_trailers`](crate::ServerConn::end_body_with_trailers)
480 /// for the fields that are refused).
481 ///
482 /// # Errors
483 ///
484 /// [`HttpError::TrailerNotAllowed`] for a forbidden field or a body
485 /// that is not chunked; the field errors of
486 /// [`Request::header`]; otherwise as [`end_body`](Self::end_body).
487 ///
488 /// # Examples
489 ///
490 /// ```
491 /// use http_net::{ClientConn, Request};
492 ///
493 /// let mut conn = ClientConn::new();
494 /// let mut out = Vec::new();
495 /// conn.send_request(&Request::put("/f").header("host", "a").chunked(), &mut out)?;
496 /// conn.send_body(b"data", &mut out)?;
497 /// conn.end_body_with_trailers(&[("content-digest", b"sha-256=:x:")], &mut out)?;
498 /// assert!(out.ends_with(b"0\r\ncontent-digest: sha-256=:x:\r\n\r\n"));
499 /// # Ok::<(), http_net::HttpError>(())
500 /// ```
501 pub fn end_body_with_trailers(
502 &mut self,
503 trailers: &[(&str, &[u8])],
504 out: &mut Vec<u8>,
505 ) -> Result<(), HttpError> {
506 if self.body_refused {
507 return Err(REFUSED);
508 }
509 let ReqWrite::Streaming(encoder) = &mut self.req else {
510 return Err(HttpError::InvalidState("no streamed request in progress"));
511 };
512 if !trailers.is_empty() && !self.req_chunked {
513 return Err(HttpError::TrailerNotAllowed);
514 }
515 if *encoder == Encoder::Chunked {
516 push_last_chunk_with_trailers(out, trailers)?;
517 } else {
518 check_trailers(trailers)?;
519 encoder.finish(out)?;
520 }
521 self.req = ReqWrite::Sent;
522 self.waiting_continue = false;
523 Ok(())
524 }
525
526 /// Advances the response parser and reports what happened.
527 ///
528 /// # Errors
529 ///
530 /// Any [`HttpError`] parse error; the connection is then closed and must
531 /// not be reused.
532 ///
533 /// # Examples
534 ///
535 /// ```
536 /// use http_net::{ClientConn, HttpError, Request};
537 ///
538 /// let mut conn = ClientConn::new();
539 /// conn.send_request(&Request::get("/").header("host", "a"), &mut Vec::new()).unwrap();
540 /// let _ = conn.feed(b"HTTP/1.1 200 OK\r\nContent-Length: 1\r\nContent-Length: 1\r\n\r\nx");
541 /// assert_eq!(conn.poll().unwrap_err(), HttpError::DuplicateContentLength);
542 /// ```
543 pub fn poll(&mut self) -> Result<ClientEvent<'_>, HttpError> {
544 loop {
545 match self.state {
546 State::Idle => {
547 if !self.buf.unread().is_empty() {
548 return Err(self.fail(HttpError::UnexpectedData));
549 }
550 if self.buf.is_eof() {
551 self.state = State::Closed;
552 return Ok(ClientEvent::Closed);
553 }
554 return Ok(ClientEvent::Idle);
555 }
556 State::Head => return self.poll_head(),
557 State::Interim => {
558 self.next_head();
559 self.state = State::Head;
560 }
561 State::Body => return self.poll_body(),
562 State::Complete => {
563 if self.reusable() {
564 self.state = State::Idle;
565 } else {
566 self.state = State::Closed;
567 return Ok(ClientEvent::Closed);
568 }
569 }
570 State::Upgraded => return Ok(ClientEvent::Upgraded),
571 State::Closed => return Ok(ClientEvent::Closed),
572 }
573 }
574 }
575
576 fn poll_head(&mut self) -> Result<ClientEvent<'_>, HttpError> {
577 loop {
578 let start = self.buf.pin();
579 let progress = self
580 .scanner
581 .advance(self.buf.from(start), &self.limits, HeadKind::Response, true)
582 .map_err(|e| self.fail(e))?;
583 let len = match progress {
584 Scan::Partial if self.buf.is_eof() => {
585 return Err(self.fail(HttpError::UnexpectedEof));
586 }
587 Scan::Partial => return Ok(ClientEvent::NeedInput),
588 Scan::Done(len) => len,
589 };
590 let info = interpret_response(
591 self.buf.from(start),
592 self.scanner.start_line(),
593 &self.scanner.fields,
594 &self.limits,
595 self.asked,
596 )
597 .map_err(|e| self.fail(e))?;
598 self.buf.consume(len);
599 self.buf.extend_pin_to_pos();
600 self.info = info;
601 let status = info.status;
602 let switched = if status == 101 {
603 if !self.upgrade_asked || !info.upgrade_field {
604 return Err(self.fail(HttpError::UnexpectedUpgrade));
605 }
606 true
607 } else {
608 self.asked.connect && (200..300).contains(&status)
609 };
610 if switched {
611 self.waiting_continue = false;
612 self.state = State::Upgraded;
613 } else if (100..200).contains(&status) {
614 // Interim response (100 Continue, 103 Early Hints): budgeted
615 // by count and by bytes, so a server cannot keep the client
616 // parsing forever without a final answer.
617 self.interim_count = self.interim_count.saturating_add(1);
618 if self.interim_count > self.limits.max_interim() {
619 return Err(self.fail(HttpError::TooManyInterim));
620 }
621 self.interim_bytes = self.interim_bytes.saturating_add(len);
622 if self.interim_bytes > self.limits.max_header_bytes() {
623 return Err(self.fail(HttpError::HeadersTooLarge));
624 }
625 if status == 100 && self.waiting_continue {
626 self.waiting_continue = false;
627 self.next_head();
628 return Ok(ClientEvent::Continue);
629 }
630 if self.report_interim {
631 self.state = State::Interim;
632 return Ok(ClientEvent::Informational(self.view()));
633 }
634 self.next_head();
635 continue;
636 } else {
637 if self.waiting_continue {
638 // RFC 9110 section 10.1.1: a final status instead of 100
639 // means the server does not want the body. A promised
640 // body that is never sent leaves the request unfinished,
641 // so the connection closes after this response.
642 self.waiting_continue = false;
643 self.body_refused = self.req != ReqWrite::Sent;
644 }
645 self.decoder.start(info.framing);
646 self.state = State::Body;
647 }
648 return Ok(ClientEvent::Response(self.view()));
649 }
650 }
651
652 /// Drops the current (interim) head and waits for the next one.
653 fn next_head(&mut self) {
654 self.buf.unpin();
655 self.buf.pin_at_pos();
656 self.scanner.reset();
657 }
658
659 fn reusable(&self) -> bool {
660 self.keep_alive && self.info.keep_alive && self.req == ReqWrite::Sent && !self.body_refused
661 }
662
663 fn poll_body(&mut self) -> Result<ClientEvent<'_>, HttpError> {
664 let mut framing_bytes = 0;
665 let step = self.decoder.step(
666 self.buf.unread(),
667 self.buf.is_eof(),
668 &self.limits,
669 &mut self.trailers,
670 self.report_trailers,
671 &mut framing_bytes,
672 );
673 self.buf.consume(framing_bytes);
674 match step.map_err(|e| self.fail(e))? {
675 Step::Data(n) => {
676 let start = self.buf.pos();
677 self.buf.consume(n);
678 Ok(ClientEvent::Body(self.buf.range(start, start + n)))
679 }
680 Step::Trailers => Ok(ClientEvent::Trailers(Headers::new(
681 self.buf.unread(),
682 &self.trailers.fields,
683 ))),
684 Step::Done => {
685 self.state = State::Complete;
686 Ok(ClientEvent::ResponseEnd)
687 }
688 Step::NeedMore => Ok(ClientEvent::NeedInput),
689 }
690 }
691
692 fn fail(&mut self, e: HttpError) -> HttpError {
693 self.state = State::Closed;
694 e
695 }
696
697 fn view(&self) -> ResponseHead<'_> {
698 ResponseHead::new(
699 self.buf.from(self.buf.pin()),
700 &self.scanner.fields,
701 &self.info,
702 )
703 }
704
705 /// The current response's head, from its arrival until the next request
706 /// is sent (still available after an upgrade).
707 ///
708 /// # Examples
709 ///
710 /// ```
711 /// use http_net::{ClientConn, ClientEvent, Request};
712 ///
713 /// let mut conn = ClientConn::new();
714 /// conn.send_request(&Request::get("/").header("host", "a"), &mut Vec::new()).unwrap();
715 /// let _ = conn.feed(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n");
716 /// while !matches!(conn.poll().unwrap(), ClientEvent::ResponseEnd) {}
717 /// assert_eq!(conn.response().unwrap().status(), 200);
718 /// ```
719 #[must_use]
720 pub fn response(&self) -> Option<ResponseHead<'_>> {
721 match self.state {
722 State::Body | State::Complete | State::Upgraded => Some(self.view()),
723 _ => None,
724 }
725 }
726
727 /// Whether the server switched protocols and the connection now belongs
728 /// to another protocol.
729 ///
730 /// # Examples
731 ///
732 /// ```
733 /// assert!(!http_net::ClientConn::new().is_upgraded());
734 /// ```
735 #[must_use]
736 pub fn is_upgraded(&self) -> bool {
737 self.state == State::Upgraded
738 }
739
740 /// Input that arrived after the last message parsed: after
741 /// [`ClientEvent::Upgraded`], the first bytes of the new protocol.
742 ///
743 /// # Examples
744 ///
745 /// ```
746 /// use http_net::{ClientConn, ClientEvent, Request};
747 ///
748 /// let mut conn = ClientConn::new();
749 /// conn.send_request(&Request::get("/").header("host", "a").upgrade("x"), &mut Vec::new()).unwrap();
750 /// let _ = conn.feed(b"HTTP/1.1 101 Switching Protocols\r\nConnection: upgrade\r\nUpgrade: x\r\n\r\nhi");
751 /// assert!(matches!(conn.poll().unwrap(), ClientEvent::Response(_)));
752 /// assert!(matches!(conn.poll().unwrap(), ClientEvent::Upgraded));
753 /// assert_eq!(conn.remaining_input(), b"hi");
754 /// ```
755 #[must_use]
756 pub fn remaining_input(&self) -> &[u8] {
757 self.buf.unread()
758 }
759
760 /// When `true`, each interim response other than `101` (for example
761 /// `103 Early Hints`) is reported as [`ClientEvent::Informational`]
762 /// before the final response; when `false` (the default) they are
763 /// skipped. Interim responses count against
764 /// [`Limits::max_interim`](crate::Limits::max_interim) either way.
765 ///
766 /// # Examples
767 ///
768 /// ```
769 /// use http_net::{ClientConn, ClientEvent, Request};
770 ///
771 /// let mut conn = ClientConn::new();
772 /// conn.set_report_informational(true);
773 /// conn.send_request(&Request::get("/").header("host", "a"), &mut Vec::new())?;
774 /// let _ = conn.feed(b"HTTP/1.1 103 Early Hints\r\nLink: </a.css>; rel=preload\r\n\r\nHTTP/1.1 204 No Content\r\n\r\n");
775 /// let ClientEvent::Informational(hints) = conn.poll()? else { panic!() };
776 /// assert_eq!(hints.status(), 103);
777 /// assert_eq!(hints.header("link"), Some(&b"</a.css>; rel=preload"[..]));
778 /// assert!(matches!(conn.poll()?, ClientEvent::Response(r) if r.status() == 204));
779 /// # Ok::<(), http_net::HttpError>(())
780 /// ```
781 pub fn set_report_informational(&mut self, report: bool) {
782 self.report_interim = report;
783 }
784
785 /// When `true`, a response's trailer fields are kept (bounded by the
786 /// trailer budgets of [`Limits`]) and reported as
787 /// [`ClientEvent::Trailers`] just before [`ClientEvent::ResponseEnd`];
788 /// when `false` (the default) they are validated and discarded.
789 ///
790 /// # Examples
791 ///
792 /// ```
793 /// use http_net::{ClientConn, ClientEvent, Request};
794 ///
795 /// let mut conn = ClientConn::new();
796 /// conn.set_report_trailers(true);
797 /// conn.send_request(&Request::get("/").header("host", "a"), &mut Vec::new())?;
798 /// let _ = conn.feed(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n2\r\nok\r\n0\r\nServer-Timing: db;dur=5\r\n\r\n");
799 /// let mut saw = false;
800 /// loop {
801 /// match conn.poll()? {
802 /// ClientEvent::Trailers(t) => saw = t.len() == 1,
803 /// ClientEvent::ResponseEnd => break,
804 /// _ => {}
805 /// }
806 /// }
807 /// assert!(saw);
808 /// # Ok::<(), http_net::HttpError>(())
809 /// ```
810 pub fn set_report_trailers(&mut self, report: bool) {
811 self.report_trailers = report;
812 }
813
814 /// Whether a request built with
815 /// [`Request::expect_continue`](crate::Request::expect_continue) is
816 /// still waiting: no `100 Continue`, no final response, and no body
817 /// bytes sent yet. A client may stop waiting on its own timer and send
818 /// the body anyway (RFC 9110 section 10.1.1).
819 ///
820 /// # Examples
821 ///
822 /// ```
823 /// use http_net::{ClientConn, ClientEvent, Request};
824 ///
825 /// let mut conn = ClientConn::new();
826 /// let mut out = Vec::new();
827 /// let upload = Request::put("/big").header("host", "a").content_length(4).expect_continue();
828 /// conn.send_request(&upload, &mut out)?;
829 /// assert!(conn.is_waiting_for_continue());
830 /// let _ = conn.feed(b"HTTP/1.1 100 Continue\r\n\r\n");
831 /// assert!(matches!(conn.poll()?, ClientEvent::Continue));
832 /// assert!(!conn.is_waiting_for_continue());
833 /// conn.send_body(b"data", &mut out)?;
834 /// conn.end_body(&mut out)?;
835 /// # Ok::<(), http_net::HttpError>(())
836 /// ```
837 #[must_use]
838 pub fn is_waiting_for_continue(&self) -> bool {
839 self.waiting_continue
840 }
841
842 /// Whether the connection can carry another request now: no request
843 /// is outstanding (or the last response completed) and both sides allow
844 /// keep-alive.
845 ///
846 /// # Examples
847 ///
848 /// ```
849 /// use http_net::{ClientConn, ClientEvent, Request};
850 ///
851 /// let mut conn = ClientConn::new();
852 /// assert!(conn.is_reusable());
853 /// conn.send_request(&Request::get("/").header("host", "a"), &mut Vec::new())?;
854 /// assert!(!conn.is_reusable());
855 /// let _ = conn.feed(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n");
856 /// while !matches!(conn.poll()?, ClientEvent::ResponseEnd) {}
857 /// assert!(conn.is_reusable());
858 /// # Ok::<(), http_net::HttpError>(())
859 /// ```
860 #[must_use]
861 pub fn is_reusable(&self) -> bool {
862 match self.state {
863 State::Idle => self.buf.unread().is_empty() && !self.buf.is_eof(),
864 State::Complete => self.reusable() && self.buf.unread().is_empty(),
865 _ => false,
866 }
867 }
868
869 /// The connection's phase, for response deadlines: [`Progress::Idle`]
870 /// with no request outstanding, [`Progress::WritingBody`] while a
871 /// request body is streamed, [`Progress::AwaitingResponse`] until the
872 /// first byte of the response, then [`Progress::Head`] and
873 /// [`Progress::Body`].
874 ///
875 /// # Examples
876 ///
877 /// ```
878 /// use http_net::{ClientConn, Progress, Request};
879 ///
880 /// let mut conn = ClientConn::new();
881 /// assert_eq!(conn.progress(), Progress::Idle);
882 /// conn.send_request(&Request::get("/").header("host", "a"), &mut Vec::new())?;
883 /// assert_eq!(conn.progress(), Progress::AwaitingResponse);
884 /// # Ok::<(), http_net::HttpError>(())
885 /// ```
886 #[must_use]
887 pub fn progress(&self) -> Progress {
888 if matches!(self.req, ReqWrite::Streaming(_))
889 && !self.body_refused
890 && !matches!(self.state, State::Closed | State::Upgraded)
891 {
892 return Progress::WritingBody;
893 }
894 match self.state {
895 State::Idle => Progress::Idle,
896 State::Head if self.buf.from(self.buf.pin()).is_empty() => Progress::AwaitingResponse,
897 State::Head | State::Interim => Progress::Head,
898 State::Body => Progress::Body,
899 State::Complete if self.reusable() => Progress::Idle,
900 State::Complete | State::Closed => Progress::Closed,
901 State::Upgraded => Progress::Upgraded,
902 }
903 }
904}
905
906#[cfg(test)]
907mod tests {
908 use super::*;
909 use crate::head::BodyFraming;
910
911 fn get() -> Request {
912 Request::get("/").header("host", "a")
913 }
914
915 #[test]
916 fn test_interim_responses_are_skipped() {
917 let mut conn = ClientConn::new();
918 conn.send_request(&get(), &mut Vec::new()).unwrap();
919 let _ = conn.feed(b"HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 103 Early Hints\r\nLink: </a>\r\n\r\nHTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok");
920 let ClientEvent::Response(r) = conn.poll().unwrap() else {
921 panic!("expected response");
922 };
923 assert_eq!(r.status(), 200);
924 assert!(matches!(conn.poll().unwrap(), ClientEvent::Body(b"ok")));
925 assert!(matches!(conn.poll().unwrap(), ClientEvent::ResponseEnd));
926 assert!(matches!(conn.poll().unwrap(), ClientEvent::Idle));
927 }
928
929 #[test]
930 fn test_unrequested_upgrade_and_unsolicited_data_are_errors() {
931 let mut conn = ClientConn::new();
932 conn.send_request(&get(), &mut Vec::new()).unwrap();
933 let _ = conn
934 .feed(b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: x\r\nConnection: upgrade\r\n\r\n");
935 assert_eq!(conn.poll().unwrap_err(), HttpError::UnexpectedUpgrade);
936
937 let mut conn = ClientConn::new();
938 let _ = conn.feed(b"HTTP/1.1 200 OK\r\n\r\n");
939 assert_eq!(conn.poll().unwrap_err(), HttpError::UnexpectedData);
940 }
941
942 #[test]
943 fn test_head_response_has_no_body_and_connection_is_reused() {
944 let mut conn = ClientConn::new();
945 conn.send_request(
946 &Request::new("HEAD", "/").header("host", "a"),
947 &mut Vec::new(),
948 )
949 .unwrap();
950 let _ = conn.feed(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n");
951 let ClientEvent::Response(r) = conn.poll().unwrap() else {
952 panic!("expected response");
953 };
954 assert_eq!(r.framing(), BodyFraming::None);
955 assert!(matches!(conn.poll().unwrap(), ClientEvent::ResponseEnd));
956 assert!(matches!(conn.poll().unwrap(), ClientEvent::Idle));
957 conn.send_request(&get(), &mut Vec::new()).unwrap();
958 assert!(matches!(
959 conn.send_request(&get(), &mut Vec::new()),
960 Err(HttpError::InvalidState(_))
961 ));
962 }
963
964 #[test]
965 fn test_close_and_streaming_request_prevent_reuse() {
966 let mut conn = ClientConn::new();
967 conn.send_request(&get().close(), &mut Vec::new()).unwrap();
968 let _ = conn.feed(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n");
969 while !matches!(conn.poll().unwrap(), ClientEvent::ResponseEnd) {}
970 assert!(matches!(conn.poll().unwrap(), ClientEvent::Closed));
971 assert!(conn.send_request(&get(), &mut Vec::new()).is_err());
972
973 let mut conn = ClientConn::new();
974 let mut out = Vec::new();
975 conn.send_request(&Request::post("/").header("host", "a").chunked(), &mut out)
976 .unwrap();
977 let _ = conn.feed(b"HTTP/1.1 413 Content Too Large\r\nContent-Length: 0\r\n\r\n");
978 while !matches!(conn.poll().unwrap(), ClientEvent::ResponseEnd) {}
979 assert!(matches!(conn.poll().unwrap(), ClientEvent::Closed));
980 assert!(matches!(conn.send_body(b"x", &mut out), Ok(())));
981 assert!(matches!(conn.end_body(&mut out), Ok(())));
982 assert!(matches!(
983 conn.end_body(&mut out),
984 Err(HttpError::InvalidState(_))
985 ));
986 }
987
988 #[test]
989 fn test_eof_before_response_is_an_error() {
990 let mut conn = ClientConn::new();
991 conn.send_request(&get(), &mut Vec::new()).unwrap();
992 conn.feed_eof();
993 assert_eq!(conn.poll().unwrap_err(), HttpError::UnexpectedEof);
994 }
995}