trillium_http/conn.rs
1use crate::{
2 Body, Buffer, Headers, HttpContext,
3 KnownHeaderName::Host,
4 Method, ProtocolSession, ReceivedBody, Status, Swansong, TypeSet, Version,
5 after_send::{AfterSend, SendStatus},
6 h2::H2Connection,
7 h3::H3Connection,
8 liveness::{CancelOnDisconnect, LivenessFut, PeerGone},
9 received_body::ReceivedBodyState,
10 util::encoding,
11};
12use encoding_rs::Encoding;
13use futures_lite::{
14 future,
15 io::{AsyncRead, AsyncWrite},
16};
17use std::{
18 borrow::Cow,
19 fmt::{self, Debug, Formatter},
20 future::Future,
21 net::IpAddr,
22 pin::pin,
23 str,
24 sync::Arc,
25 time::Instant,
26};
27mod h1;
28#[cfg(test)]
29mod h1_tests;
30mod h2;
31mod h3;
32mod shared;
33pub(crate) use h1::{HeadError, write_headers_or_trailers};
34pub(crate) use h3::H3FirstFrame;
35pub(crate) use shared::ConnParts;
36
37/// An HTTP connection.
38///
39/// This struct represents both the request and the response, and holds the
40/// transport over which the response will be sent.
41#[derive(fieldwork::Fieldwork)]
42pub struct Conn<Transport> {
43 #[field(get)]
44 /// the shared [`HttpContext`]
45 pub(crate) context: Arc<HttpContext>,
46
47 /// request [headers](Headers)
48 #[field(get, get_mut)]
49 pub(crate) request_headers: Headers,
50
51 /// The previous request's header map, emptied but with its capacity retained. HTTP/1.x
52 /// parses the next request into it, salvaging entries from the current `request_headers`
53 /// as it goes; see `Headers::extend_parse_reusing`. `None` until a connection's second
54 /// request, so single-request connections never allocate it.
55 pub(crate) spare_request_headers: Option<Headers>,
56
57 /// response [headers](Headers)
58 #[field(get, get_mut)]
59 pub(crate) response_headers: Headers,
60
61 pub(crate) path: Cow<'static, str>,
62
63 /// the http method for this conn's request
64 ///
65 /// ```
66 /// # use trillium_http::{Conn, Method};
67 /// let mut conn = Conn::new_synthetic(Method::Get, "/some/path?and&a=query", ());
68 /// assert_eq!(conn.method(), Method::Get);
69 /// ```
70 #[field(get, set, copy)]
71 pub(crate) method: Method,
72
73 /// the http status for this conn, if set
74 #[field(get, copy)]
75 pub(crate) status: Option<Status>,
76
77 /// The HTTP protocol version in use on this connection.
78 ///
79 /// ```
80 /// # use trillium_http::{Conn, Method, Version};
81 /// let conn = Conn::new_synthetic(Method::Get, "/", ());
82 /// assert_eq!(conn.http_version(), Version::Http1_1);
83 /// ```
84 #[field(get = http_version, copy)]
85 pub(crate) version: Version,
86
87 /// the [state typemap](TypeSet) for this conn
88 #[field(get, get_mut)]
89 pub(crate) state: TypeSet,
90
91 /// the response [body](Body)
92 ///
93 /// ```
94 /// # use trillium_testing::HttpTest;
95 /// HttpTest::new(|conn| async move { conn.with_response_body("hello") })
96 /// .get("/")
97 /// .block()
98 /// .assert_body("hello");
99 ///
100 /// HttpTest::new(|conn| async move { conn.with_response_body(String::from("world")) })
101 /// .get("/")
102 /// .block()
103 /// .assert_body("world");
104 ///
105 /// HttpTest::new(|conn| async move { conn.with_response_body(vec![99, 97, 116]) })
106 /// .get("/")
107 /// .block()
108 /// .assert_body("cat");
109 /// ```
110 #[field(get, set, into, option_set_some, take, with)]
111 pub(crate) response_body: Option<Body>,
112
113 /// the transport
114 ///
115 /// This should only be used to call your own custom methods on the transport that do not read
116 /// or write any data. Calling any method that reads from or writes to the transport will
117 /// disrupt the HTTP protocol. If you're looking to transition from HTTP to another protocol,
118 /// use an HTTP upgrade.
119 #[field(get, get_mut)]
120 pub(crate) transport: Transport,
121
122 pub(crate) buffer: Buffer,
123
124 pub(crate) request_body_state: ReceivedBodyState,
125
126 pub(crate) after_send: AfterSend,
127
128 /// whether the connection is secure
129 ///
130 /// note that this does not necessarily indicate that the transport itself is secure, as it may
131 /// indicate that `trillium_http` is behind a trusted reverse proxy that has terminated tls and
132 /// provided appropriate headers to indicate this.
133 #[field(get, set, rename_predicates)]
134 pub(crate) secure: bool,
135
136 /// The [`Instant`] that the first header bytes for this conn were
137 /// received, before any processing or parsing has been performed.
138 #[field(get, copy)]
139 pub(crate) start_time: Instant,
140
141 /// The IP Address for the connection, if available
142 #[field(set, get, copy, into)]
143 pub(crate) peer_ip: Option<IpAddr>,
144
145 /// the `:authority` pseudo-header
146 #[field(set, get, into)]
147 pub(crate) authority: Option<Cow<'static, str>>,
148
149 /// the `:scheme` pseudo-header
150 #[field(set, get, into)]
151 pub(crate) scheme: Option<Cow<'static, str>>,
152
153 /// the [`ProtocolSession`] for this conn — the per-protocol session state
154 /// (h2/h3 connection driver and stream id) bundled into a single enum so the
155 /// "set together" invariant is enforced at the type level. `Http1` for
156 /// h1 / synthetic conns.
157 pub(crate) protocol_session: ProtocolSession,
158
159 /// the `:protocol` pseudo-header (extended CONNECT)
160 #[field(set, get, into)]
161 pub(crate) protocol: Option<Cow<'static, str>>,
162
163 /// request trailers, populated after the request body has been fully read
164 #[field(get, get_mut)]
165 pub(crate) request_trailers: Option<Headers>,
166
167 /// Marker set via [`Conn::upgrade`].
168 pub(crate) upgrade: bool,
169
170 /// Resolves when the peer abandons this stream. HTTP/3 only — h1 detects departure by
171 /// reading and h2 through its connection driver, but h3's QUIC signals are invisible at
172 /// the transport seam, so the runtime adapter supplies this. See [`PeerGone`].
173 #[field = false]
174 pub(crate) peer_gone: Option<PeerGone>,
175}
176
177impl<Transport> Debug for Conn<Transport> {
178 fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
179 f.debug_struct("Conn")
180 .field("context", &self.context)
181 .field("request_headers", &self.request_headers)
182 .field("spare_request_headers", &format_args!(".."))
183 .field("response_headers", &self.response_headers)
184 .field("path", &self.path)
185 .field("method", &self.method)
186 .field("status", &self.status)
187 .field("version", &self.version)
188 .field("state", &self.state)
189 .field("response_body", &self.response_body)
190 .field("transport", &format_args!(".."))
191 .field("buffer", &format_args!(".."))
192 .field("request_body_state", &self.request_body_state)
193 .field("secure", &self.secure)
194 .field("after_send", &format_args!(".."))
195 .field("start_time", &self.start_time)
196 .field("peer_ip", &self.peer_ip)
197 .field("authority", &self.authority)
198 .field("scheme", &self.scheme)
199 .field("protocol", &self.protocol)
200 .field("protocol_session", &self.protocol_session)
201 .field("request_trailers", &self.request_trailers)
202 .field("upgrade", &self.upgrade)
203 .field("peer_gone", &format_args!(".."))
204 .finish()
205 }
206}
207
208impl<Transport> Conn<Transport>
209where
210 Transport: AsyncRead + AsyncWrite + Unpin + Send + Sync + 'static,
211{
212 /// Returns the shared state typemap for this conn.
213 pub fn shared_state(&self) -> &TypeSet {
214 &self.context.shared_state
215 }
216
217 /// sets the http status code from any `TryInto<Status>`.
218 ///
219 /// ```
220 /// # use trillium_http::Status;
221 /// # trillium_testing::HttpTest::new(|mut conn| async move {
222 /// assert!(conn.status().is_none());
223 ///
224 /// conn.set_status(200); // a status can be set as a u16
225 /// assert_eq!(conn.status().unwrap(), Status::Ok);
226 ///
227 /// conn.set_status(Status::ImATeapot); // or as a Status
228 /// assert_eq!(conn.status().unwrap(), Status::ImATeapot);
229 /// conn
230 /// # }).get("/").block().assert_status(Status::ImATeapot);
231 /// ```
232 pub fn set_status(&mut self, status: impl TryInto<Status>) -> &mut Self {
233 self.status = Some(status.try_into().unwrap_or_else(|_| {
234 log::error!("attempted to set an invalid status code");
235 Status::InternalServerError
236 }));
237 self
238 }
239
240 /// sets the http status code from any `TryInto<Status>`, returning Conn
241 #[must_use]
242 pub fn with_status(mut self, status: impl TryInto<Status>) -> Self {
243 self.set_status(status);
244 self
245 }
246
247 /// The status to send on the wire: the explicitly-set status, or a
248 /// method-appropriate default when a handler left it unset. Unhandled
249 /// requests default to `404 Not Found`, except CONNECT, which defaults to
250 /// `501 Not Implemented`: an origin server implements no tunnel, and 404's
251 /// resource model does not apply to CONNECT's authority-form target.
252 pub(crate) fn response_status(&self) -> Status {
253 self.status.unwrap_or(match self.method {
254 Method::Connect => Status::NotImplemented,
255 _ => Status::NotFound,
256 })
257 }
258
259 /// retrieves the path part of the request url, up to and excluding any query component
260 /// ```
261 /// # use trillium_testing::HttpTest;
262 /// HttpTest::new(|mut conn| async move {
263 /// assert_eq!(conn.path(), "/some/path");
264 /// conn.with_status(200)
265 /// })
266 /// .get("/some/path?and&a=query")
267 /// .block()
268 /// .assert_ok();
269 /// ```
270 pub fn path(&self) -> &str {
271 match self.path.split_once('?') {
272 Some((path, _)) => path,
273 None => &self.path,
274 }
275 }
276
277 /// retrieves the combined path and any query
278 pub fn path_and_query(&self) -> &str {
279 &self.path
280 }
281
282 /// retrieves the query component of the path, or an empty &str
283 ///
284 /// ```
285 /// # use trillium_testing::HttpTest;
286 /// let server = HttpTest::new(|conn| async move {
287 /// let querystring = conn.querystring().to_string();
288 /// conn.with_response_body(querystring).with_status(200)
289 /// });
290 ///
291 /// server
292 /// .get("/some/path?and&a=query")
293 /// .block()
294 /// .assert_body("and&a=query");
295 ///
296 /// server.get("/some/path").block().assert_body("");
297 /// ```
298 pub fn querystring(&self) -> &str {
299 self.path
300 .split_once('?')
301 .map(|(_, query)| query)
302 .unwrap_or_default()
303 }
304
305 /// get the host for this conn, if it exists.
306 ///
307 /// On protocol versions where the equivalent of `Host` is `:authority`, this returns
308 /// `:authority`.
309 pub fn host(&self) -> Option<&str> {
310 self.request_headers
311 .get_str(Host)
312 .or_else(|| self.authority())
313 }
314
315 /// set the host for this conn
316 pub fn set_host(&mut self, host: String) -> &mut Self {
317 self.request_headers.insert(Host, host);
318 self
319 }
320
321 /// Cancels and drops the future if the peer abandons this request
322 ///
323 /// If the client disconnects, this function will return None. If the future completes
324 /// without disconnection, this future will return Some containing the output of the future.
325 ///
326 /// See [`is_disconnected`][Self::is_disconnected] for how departure is detected on each
327 /// protocol. On HTTP/1.x, where detection is by reading, any bytes the client sends while
328 /// the future runs — an unread request body, or pipelined requests — are buffered, up to
329 /// 16kb. A client that fills that allowance is considered alive for the remainder of the
330 /// future, even if it disconnects afterwards. If the request has a body, read it before
331 /// calling this.
332 ///
333 /// Note that the inner future cannot borrow conn, so you will need to clone or take any
334 /// information needed to execute the future prior to executing this method.
335 ///
336 /// # Example
337 ///
338 /// ```rust
339 /// # use futures_lite::{AsyncRead, AsyncWrite};
340 /// # use trillium_http::{Conn, Method};
341 /// async fn something_slow_and_cancel_safe() -> String {
342 /// String::from("this was not actually slow")
343 /// }
344 /// async fn handler<T>(mut conn: Conn<T>) -> Conn<T>
345 /// where
346 /// T: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
347 /// {
348 /// let Some(returned_body) = conn
349 /// .cancel_on_disconnect(async { something_slow_and_cancel_safe().await })
350 /// .await
351 /// else {
352 /// return conn;
353 /// };
354 /// conn.with_response_body(returned_body).with_status(200)
355 /// }
356 /// ```
357 pub async fn cancel_on_disconnect<'a, Fut>(&'a mut self, fut: Fut) -> Option<Fut::Output>
358 where
359 Fut: Future + Send + 'a,
360 {
361 CancelOnDisconnect(self, pin!(fut)).await
362 }
363
364 /// Check whether the peer has abandoned this request.
365 ///
366 /// How departure becomes observable is protocol-specific. On HTTP/1.x this reads the
367 /// transport: any bytes the client sends — an unread request body, or pipelined requests —
368 /// are buffered, up to 16kb, and count as evidence of liveness. A client that fills that
369 /// allowance is reported as connected until the buffered bytes are read, so read the request
370 /// body before polling this in a long-running handler. On HTTP/2 this observes stream reset
371 /// and connection teardown, and on HTTP/3 stream cancellation and connection loss; neither
372 /// reads the transport, and neither treats the peer's half-close — which both protocols do
373 /// as a matter of course once the request is complete — as a departure.
374 ///
375 /// A client that vanishes without signalling is only reported once the transport notices,
376 /// which on HTTP/3 is QUIC's negotiated idle timeout and on HTTP/1.x and HTTP/2 depends on
377 /// the TCP configuration.
378 ///
379 /// # Example
380 ///
381 /// This is best to use at appropriate points in a long-running handler, like:
382 ///
383 /// ```rust
384 /// # use futures_lite::{AsyncRead, AsyncWrite};
385 /// # use trillium_http::{Conn, Method};
386 /// # async fn something_slow_but_not_cancel_safe() {}
387 /// async fn handler<T>(mut conn: Conn<T>) -> Conn<T>
388 /// where
389 /// T: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
390 /// {
391 /// for _ in 0..100 {
392 /// if conn.is_disconnected().await {
393 /// return conn;
394 /// }
395 /// something_slow_but_not_cancel_safe().await;
396 /// }
397 /// conn.with_status(200)
398 /// }
399 /// ```
400 pub async fn is_disconnected(&mut self) -> bool {
401 future::poll_once(LivenessFut::new(self)).await.is_some()
402 }
403
404 /// returns the [`encoding_rs::Encoding`] for this request, as determined from the mime-type
405 /// charset, if available
406 ///
407 /// ```
408 /// # use trillium_testing::HttpTest;
409 /// HttpTest::new(|mut conn| async move {
410 /// assert_eq!(conn.request_encoding(), encoding_rs::UTF_8); // the default
411 ///
412 /// conn.request_headers_mut()
413 /// .insert("content-type", "text/plain;charset=utf-16");
414 /// assert_eq!(conn.request_encoding(), encoding_rs::UTF_16LE);
415 ///
416 /// conn.with_status(200)
417 /// })
418 /// .get("/")
419 /// .block()
420 /// .assert_ok();
421 /// ```
422 pub fn request_encoding(&self) -> &'static Encoding {
423 encoding(&self.request_headers)
424 }
425
426 /// returns the [`encoding_rs::Encoding`] for this response, as
427 /// determined from the mime-type charset, if available
428 ///
429 /// ```
430 /// # use trillium_testing::HttpTest;
431 /// HttpTest::new(|mut conn| async move {
432 /// assert_eq!(conn.response_encoding(), encoding_rs::UTF_8); // the default
433 /// conn.response_headers_mut()
434 /// .insert("content-type", "text/plain;charset=utf-16");
435 ///
436 /// assert_eq!(conn.response_encoding(), encoding_rs::UTF_16LE);
437 ///
438 /// conn.with_status(200)
439 /// })
440 /// .get("/")
441 /// .block()
442 /// .assert_ok();
443 /// ```
444 pub fn response_encoding(&self) -> &'static Encoding {
445 encoding(&self.response_headers)
446 }
447
448 /// returns a [`ReceivedBody`] that references this conn. the conn
449 /// retains all data and holds the singular transport, but the
450 /// `ReceivedBody` provides an interface to read body content.
451 ///
452 /// If the request included an `Expect: 100-continue` header, the 100 Continue response is sent
453 /// lazily on the first read from the returned [`ReceivedBody`].
454 /// ```
455 /// # use trillium_testing::HttpTest;
456 /// let server = HttpTest::new(|mut conn| async move {
457 /// let request_body = conn.request_body();
458 /// assert_eq!(request_body.content_length(), Some(5));
459 /// assert_eq!(request_body.read_string().await.unwrap(), "hello");
460 /// conn.with_status(200)
461 /// });
462 ///
463 /// server.post("/").with_body("hello").block().assert_ok();
464 /// ```
465 pub fn request_body(&mut self) -> ReceivedBody<'_, Transport> {
466 let needs_100_continue = self.needs_100_continue();
467 let body = self.build_request_body();
468 if needs_100_continue {
469 body.with_send_100_continue()
470 } else {
471 body
472 }
473 }
474
475 /// returns a clone of the [`swansong::Swansong`] for this Conn. use
476 /// this to gracefully stop long-running futures and streams
477 /// inside of handler functions
478 pub fn swansong(&self) -> Swansong {
479 self.protocol_session
480 .h3_connection()
481 .map_or_else(|| self.context.swansong.clone(), |h| h.swansong().clone())
482 }
483
484 /// Registers a function to call after the http response has been
485 /// completely transferred.
486 ///
487 /// The callback is guaranteed to fire **exactly once** before the conn is
488 /// dropped. Either the codec's send path invokes it with the real outcome,
489 /// or — if the conn is dropped before send completes (handler panic,
490 /// transport error, mid-write disconnect) — the drop fallback invokes it
491 /// with a `SendStatus` whose `is_success()` returns false. Multiple
492 /// registrations on the same conn chain in registration order.
493 ///
494 /// Because firing is ordered by send-completion rather than handler return,
495 /// this is the right hook for instrumentation that wants to report what the
496 /// peer actually observed.
497 ///
498 /// This is a sync function and should be computationally lightweight. If
499 /// your _application_ needs additional async processing, use your runtime's
500 /// task spawn within this hook. If your _library_ needs additional async
501 /// processing in an `after_send` hook, please open an issue.
502 pub fn after_send<F>(&mut self, after_send: F)
503 where
504 F: FnOnce(SendStatus) + Send + Sync + 'static,
505 {
506 self.after_send.append(after_send);
507 }
508
509 /// applies a mapping function from one transport to another. This
510 /// is particularly useful for boxing the transport. unless you're
511 /// sure this is what you're looking for, you probably don't want
512 /// to be using this
513 pub fn map_transport<NewTransport>(
514 self,
515 f: impl Fn(Transport) -> NewTransport,
516 ) -> Conn<NewTransport>
517 where
518 NewTransport: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
519 {
520 // Manual respread: rustc treats `Conn<Transport>` and `Conn<NewTransport>` as
521 // disjoint types and rejects `..self` without the unstable
522 // `type_changing_struct_update` feature. If a new field is added to `Conn`,
523 // update this respread, `Upgrade::map_transport`, and `From<Conn> for Upgrade`
524 // (`upgrade.rs`) — they share this drift hazard.
525 Conn {
526 context: self.context,
527 request_headers: self.request_headers,
528 spare_request_headers: self.spare_request_headers,
529 response_headers: self.response_headers,
530 method: self.method,
531 response_body: self.response_body,
532 path: self.path,
533 status: self.status,
534 version: self.version,
535 state: self.state,
536 transport: f(self.transport),
537 buffer: self.buffer,
538 request_body_state: self.request_body_state,
539 secure: self.secure,
540 after_send: self.after_send,
541 start_time: self.start_time,
542 peer_ip: self.peer_ip,
543 authority: self.authority,
544 scheme: self.scheme,
545 protocol: self.protocol,
546 protocol_session: self.protocol_session,
547 request_trailers: self.request_trailers,
548 upgrade: self.upgrade,
549 peer_gone: self.peer_gone,
550 }
551 }
552
553 /// whether this conn is suitable for an http upgrade to another protocol
554 pub fn should_upgrade(&self) -> bool {
555 self.upgrade
556 || (self.method() == Method::Connect && self.status == Some(Status::Ok))
557 || self.status == Some(Status::SwitchingProtocols)
558 }
559
560 /// Mark this conn to be handed off as an upgrade once the response headers are sent.
561 /// Set the response status (typically `200`) and any headers describing the upgraded
562 /// byte stream before calling; the handler's `upgrade` method receives an [`Upgrade`]
563 /// with per-protocol framing applied on its `AsyncRead`/`AsyncWrite`.
564 #[doc(hidden)]
565 #[must_use]
566 pub fn upgrade(mut self) -> Self {
567 self.upgrade = true;
568 self
569 }
570
571 #[doc(hidden)]
572 pub fn finalize_headers(&mut self) {
573 if self.version == Version::Http3 {
574 self.finalize_response_headers_h3();
575 } else {
576 self.finalize_response_headers_1x();
577 }
578 }
579
580 /// the [`H2Connection`] driver for this conn, if this is an HTTP/2 request
581 pub fn h2_connection(&self) -> Option<&Arc<H2Connection>> {
582 self.protocol_session.h2_connection()
583 }
584
585 /// the h2 stream id for this conn, if this is an HTTP/2 request
586 pub fn h2_stream_id(&self) -> Option<u32> {
587 self.protocol_session.h2_stream_id()
588 }
589
590 /// the [`H3Connection`] driver for this conn, if this is an HTTP/3 request
591 pub fn h3_connection(&self) -> Option<&Arc<H3Connection>> {
592 self.protocol_session.h3_connection()
593 }
594
595 /// the h3 stream id for this conn, if this is an HTTP/3 request
596 pub fn h3_stream_id(&self) -> Option<u64> {
597 self.protocol_session.h3_stream_id()
598 }
599}