hyper/server/conn/
http1.rs

1//! HTTP/1 Server Connections.
2
3use std::error::Error as StdError;
4use std::fmt;
5use std::future::Future;
6use std::pin::Pin;
7use std::sync::Arc;
8use std::task::{Context, Poll};
9use std::time::Duration;
10
11use crate::rt::{Read, Write};
12use crate::upgrade::Upgraded;
13use bytes::Bytes;
14use futures_core::ready;
15use httparse::ParserConfig;
16
17use crate::body::{Body, Incoming as IncomingBody};
18use crate::proto;
19use crate::service::HttpService;
20use crate::{
21    common::time::{Dur, Time},
22    rt::Timer,
23};
24
25type Http1Dispatcher<T, B, S> = proto::h1::Dispatcher<
26    proto::h1::dispatch::Server<S, IncomingBody>,
27    B,
28    T,
29    proto::ServerTransaction,
30>;
31
32pin_project_lite::pin_project! {
33    /// A [`Future`](core::future::Future) representing an HTTP/1 connection, bound to a
34    /// [`Service`](crate::service::Service), returned from
35    /// [`Builder::serve_connection`](struct.Builder.html#method.serve_connection).
36    ///
37    /// To drive HTTP on this connection this future **must be polled**, typically with
38    /// `.await`. If it isn't polled, no progress will be made on this connection.
39    #[must_use = "futures do nothing unless polled"]
40    pub struct Connection<T, S>
41    where
42        S: HttpService<IncomingBody>,
43    {
44        conn: Http1Dispatcher<T, S::ResBody, S>,
45    }
46}
47
48/// A configuration builder for HTTP/1 server connections.
49///
50/// **Note**: The default values of options are *not considered stable*. They
51/// are subject to change at any time.
52///
53/// # Example
54///
55/// ```
56/// # use std::time::Duration;
57/// # use hyper::server::conn::http1::Builder;
58/// # fn main() {
59/// let mut http = Builder::new();
60/// // Set options one at a time
61/// http.half_close(false);
62///
63/// // Or, chain multiple options
64/// http.keep_alive(false).title_case_headers(true).max_buf_size(8192);
65///
66/// # }
67/// ```
68///
69/// Use [`Builder::serve_connection`](struct.Builder.html#method.serve_connection)
70/// to bind the built connection to a service.
71#[derive(Clone, Debug)]
72pub struct Builder {
73    h1_parser_config: ParserConfig,
74    timer: Time,
75    h1_half_close: bool,
76    h1_keep_alive: bool,
77    h1_title_case_headers: bool,
78    h1_preserve_header_case: bool,
79    h1_max_headers: Option<usize>,
80    h1_max_header_size: Option<usize>,
81    h1_header_read_timeout: Dur,
82    h1_writev: Option<bool>,
83    max_buf_size: Option<usize>,
84    pipeline_flush: bool,
85    date_header: bool,
86}
87
88/// Deconstructed parts of a `Connection`.
89///
90/// This allows taking apart a `Connection` at a later time, in order to
91/// reclaim the IO object, and additional related pieces.
92#[derive(Debug)]
93#[non_exhaustive]
94pub struct Parts<T, S> {
95    /// The original IO object used in the handshake.
96    pub io: T,
97    /// A buffer of bytes that have been read but not processed as HTTP.
98    ///
99    /// If the client sent additional bytes after its last request, and
100    /// this connection "ended" with an upgrade, the read buffer will contain
101    /// those bytes.
102    ///
103    /// You will want to check for any existing bytes if you plan to continue
104    /// communicating on the IO object.
105    pub read_buf: Bytes,
106    /// The `Service` used to serve this connection.
107    pub service: S,
108}
109
110// ===== impl Connection =====
111
112impl<I, S> fmt::Debug for Connection<I, S>
113where
114    S: HttpService<IncomingBody>,
115{
116    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
117        f.debug_struct("Connection").finish()
118    }
119}
120
121impl<I, B, S> Connection<I, S>
122where
123    S: HttpService<IncomingBody, ResBody = B>,
124    S::Error: Into<Box<dyn StdError + Send + Sync>>,
125    I: Read + Write + Unpin,
126    B: Body + 'static,
127    B::Error: Into<Box<dyn StdError + Send + Sync>>,
128{
129    /// Start a graceful shutdown process for this connection.
130    ///
131    /// This `Connection` should continue to be polled until shutdown
132    /// can finish.
133    ///
134    /// # Note
135    ///
136    /// This should only be called while the `Connection` future is still
137    /// pending. If called after `Connection::poll` has resolved, this does
138    /// nothing.
139    pub fn graceful_shutdown(mut self: Pin<&mut Self>) {
140        self.conn.disable_keep_alive();
141    }
142
143    /// Return the inner IO object, and additional information.
144    ///
145    /// If the IO object has been "rewound" the io will not contain those bytes rewound.
146    /// This should only be called after `poll_without_shutdown` signals
147    /// that the connection is "done". Otherwise, it may not have finished
148    /// flushing all necessary HTTP bytes.
149    ///
150    /// # Panics
151    /// This method will panic if this connection is using an h2 protocol.
152    pub fn into_parts(self) -> Parts<I, S> {
153        let (io, read_buf, dispatch) = self.conn.into_inner();
154        Parts {
155            io,
156            read_buf,
157            service: dispatch.into_service(),
158        }
159    }
160
161    /// Poll the connection for completion, but without calling `shutdown`
162    /// on the underlying IO.
163    ///
164    /// This is useful to allow running a connection while doing an HTTP
165    /// upgrade. Once the upgrade is completed, the connection would be "done",
166    /// but it is not desired to actually shutdown the IO object. Instead you
167    /// would take it back using `into_parts`.
168    pub fn poll_without_shutdown(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<()>>
169    where
170        S: Unpin,
171        S::Future: Unpin,
172    {
173        self.conn.poll_without_shutdown(cx)
174    }
175
176    /// Prevent shutdown of the underlying IO object at the end of service the request,
177    /// instead run `into_parts`. This is a convenience wrapper over `poll_without_shutdown`.
178    ///
179    /// # Error
180    ///
181    /// This errors if the underlying connection protocol is not HTTP/1.
182    #[allow(clippy::missing_panics_doc)]
183    pub fn without_shutdown(self) -> impl Future<Output = crate::Result<Parts<I, S>>> {
184        let mut zelf = Some(self);
185        crate::common::future::poll_fn(move |cx| {
186            ready!(zelf
187                .as_mut()
188                .expect("server connection polled after completion")
189                .conn
190                .poll_without_shutdown(cx))?;
191            Poll::Ready(Ok(zelf
192                .take()
193                .expect("server connection missing before completion")
194                .into_parts()))
195        })
196    }
197
198    /// Enable this connection to support higher-level HTTP upgrades.
199    ///
200    /// See [the `upgrade` module](crate::upgrade) for more.
201    pub fn with_upgrades(self) -> UpgradeableConnection<I, S>
202    where
203        I: Send,
204    {
205        UpgradeableConnection { inner: Some(self) }
206    }
207}
208
209impl<I, B, S> Future for Connection<I, S>
210where
211    S: HttpService<IncomingBody, ResBody = B>,
212    S::Error: Into<Box<dyn StdError + Send + Sync>>,
213    I: Read + Write + Unpin,
214    B: Body + 'static,
215    B::Error: Into<Box<dyn StdError + Send + Sync>>,
216{
217    type Output = crate::Result<()>;
218
219    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
220        match ready!(Pin::new(&mut self.conn).poll(cx)) {
221            Ok(done) => {
222                match done {
223                    proto::Dispatched::Shutdown => {}
224                    proto::Dispatched::Upgrade(pending) => {
225                        // With no `Send` bound on `I`, we can't try to do
226                        // upgrades here. In case a user was trying to use
227                        // `Body::on_upgrade` with this API, send a special
228                        // error letting them know about that.
229                        pending.manual();
230                    }
231                }
232                Poll::Ready(Ok(()))
233            }
234            Err(e) => Poll::Ready(Err(e)),
235        }
236    }
237}
238
239// ===== impl Builder =====
240
241impl Builder {
242    /// Create a new connection builder.
243    pub fn new() -> Self {
244        Self {
245            h1_parser_config: ParserConfig::default(),
246            timer: Time::Empty,
247            h1_half_close: false,
248            h1_keep_alive: true,
249            h1_title_case_headers: false,
250            h1_preserve_header_case: false,
251            h1_max_headers: None,
252            h1_max_header_size: None,
253            h1_header_read_timeout: Dur::Default(Some(Duration::from_secs(30))),
254            h1_writev: None,
255            max_buf_size: None,
256            pipeline_flush: false,
257            date_header: true,
258        }
259    }
260    /// Set whether HTTP/1 connections should support half-closures.
261    ///
262    /// Clients can chose to shutdown their write-side while waiting
263    /// for the server to respond. Setting this to `true` will
264    /// prevent closing the connection immediately if `read`
265    /// detects an EOF in the middle of a request.
266    ///
267    /// Default is `false`.
268    pub fn half_close(&mut self, val: bool) -> &mut Self {
269        self.h1_half_close = val;
270        self
271    }
272
273    /// Enables or disables HTTP/1 keep-alive.
274    ///
275    /// Default is `true`.
276    pub fn keep_alive(&mut self, val: bool) -> &mut Self {
277        self.h1_keep_alive = val;
278        self
279    }
280
281    /// Set whether HTTP/1 connections will write header names as title case at
282    /// the socket level.
283    ///
284    /// Default is `false`.
285    pub fn title_case_headers(&mut self, enabled: bool) -> &mut Self {
286        self.h1_title_case_headers = enabled;
287        self
288    }
289
290    /// Set whether multiple spaces are allowed as delimiters in request lines.
291    ///
292    /// Default is `false`.
293    pub fn allow_multiple_spaces_in_request_line_delimiters(&mut self, enabled: bool) -> &mut Self {
294        self.h1_parser_config
295            .allow_multiple_spaces_in_request_line_delimiters(enabled);
296        self
297    }
298
299    /// Set whether HTTP/1 connections will silently ignored malformed header lines.
300    ///
301    /// If this is enabled and a header line does not start with a valid header
302    /// name, or does not include a colon at all, the line will be silently ignored
303    /// and no error will be reported.
304    ///
305    /// Default is `false`.
306    pub fn ignore_invalid_headers(&mut self, enabled: bool) -> &mut Builder {
307        self.h1_parser_config
308            .ignore_invalid_headers_in_requests(enabled);
309        self
310    }
311
312    /// Set whether to support preserving original header cases.
313    ///
314    /// Currently, this will record the original cases received, and store them
315    /// in a private extension on the `Request`. It will also look for and use
316    /// such an extension in any provided `Response`.
317    ///
318    /// Since the relevant extension is still private, there is no way to
319    /// interact with the original cases. The only effect this can have now is
320    /// to forward the cases in a proxy-like fashion.
321    ///
322    /// Default is `false`.
323    pub fn preserve_header_case(&mut self, enabled: bool) -> &mut Self {
324        self.h1_preserve_header_case = enabled;
325        self
326    }
327
328    /// Set the maximum number of headers.
329    ///
330    /// When a request is received, the parser will reserve a buffer to store headers for optimal
331    /// performance.
332    ///
333    /// If server receives more headers than the buffer size, it responds to the client with
334    /// "431 Request Header Fields Too Large".
335    ///
336    /// Note that headers is allocated on the stack by default, which has higher performance. After
337    /// setting this value, headers will be allocated in heap memory, that is, heap memory
338    /// allocation will occur for each request, and there will be a performance drop of about 5%.
339    ///
340    /// Default is 100.
341    pub fn max_headers(&mut self, val: usize) -> &mut Self {
342        self.h1_max_headers = Some(val);
343        self
344    }
345
346    /// Set the maximum size of request headers (including the start line) in bytes.
347    ///
348    /// If the client sends headers exceeding this limit, the server responds to the
349    /// client with "431 Request Header Fields Too Large" and closes the connection.
350    ///
351    /// If not configured, then the [`max_buf_size`](Builder::max_buf_size) will naturally be reached and applied.
352    ///
353    /// This value is also used as the maximum size limit for chunked trailers.
354    ///
355    /// Default is `None`.
356    pub fn max_header_size(&mut self, val: usize) -> &mut Self {
357        self.h1_max_header_size = Some(val);
358        self
359    }
360
361    /// Set a timeout for reading client request headers. If a client does not
362    /// transmit the entire header within this time, the connection is closed.
363    ///
364    /// Requires a [`Timer`] set by [`Builder::timer`] to take effect. Panics if `header_read_timeout` is configured
365    /// without a [`Timer`].
366    ///
367    /// Pass `None` to disable.
368    ///
369    /// Default is 30 seconds.
370    pub fn header_read_timeout(&mut self, read_timeout: impl Into<Option<Duration>>) -> &mut Self {
371        self.h1_header_read_timeout = Dur::Configured(read_timeout.into());
372        self
373    }
374
375    /// Set whether HTTP/1 connections should try to use vectored writes,
376    /// or always flatten into a single buffer.
377    ///
378    /// Note that setting this to false may mean more copies of body data,
379    /// but may also improve performance when an IO transport doesn't
380    /// support vectored writes well, such as most TLS implementations.
381    ///
382    /// Setting this to true will force hyper to use queued strategy,
383    /// which may eliminate unnecessary cloning on some TLS backends.
384    ///
385    /// Default is `auto`. In this mode hyper will try to guess which
386    /// mode to use.
387    pub fn writev(&mut self, val: bool) -> &mut Self {
388        self.h1_writev = Some(val);
389        self
390    }
391
392    /// Set the maximum buffer size for the connection.
393    ///
394    /// Default is ~400kb.
395    ///
396    /// # Panics
397    ///
398    /// The minimum value allowed is 8192. This method panics if the passed `max` is less than the minimum.
399    pub fn max_buf_size(&mut self, max: usize) -> &mut Self {
400        assert!(
401            max >= proto::h1::MINIMUM_MAX_BUFFER_SIZE,
402            "the max_buf_size cannot be smaller than the minimum that h1 specifies."
403        );
404        self.max_buf_size = Some(max);
405        self
406    }
407
408    /// Set whether the `date` header should be included in HTTP responses.
409    ///
410    /// Note that including the `date` header is recommended by RFC 7231.
411    ///
412    /// Default is `true`.
413    pub fn auto_date_header(&mut self, enabled: bool) -> &mut Self {
414        self.date_header = enabled;
415        self
416    }
417
418    /// Aggregates flushes to better support pipelined responses.
419    ///
420    /// Experimental, may have bugs.
421    ///
422    /// Default is `false`.
423    pub fn pipeline_flush(&mut self, enabled: bool) -> &mut Self {
424        self.pipeline_flush = enabled;
425        self
426    }
427
428    /// Set the timer used in background tasks.
429    pub fn timer<M>(&mut self, timer: M) -> &mut Self
430    where
431        M: Timer + Send + Sync + 'static,
432    {
433        self.timer = Time::Timer(Arc::new(timer));
434        self
435    }
436
437    /// Bind a connection together with a [`Service`](crate::service::Service).
438    ///
439    /// This returns a Future that must be polled in order for HTTP to be
440    /// driven on the connection.
441    ///
442    /// # Panics
443    ///
444    /// If a timeout option has been configured, but a `timer` has not been
445    /// provided, calling `serve_connection` will panic.
446    ///
447    /// # Example
448    ///
449    /// ```
450    /// # use hyper::{body::Incoming, Request, Response};
451    /// # use hyper::service::Service;
452    /// # use hyper::server::conn::http1::Builder;
453    /// # use hyper::rt::{Read, Write};
454    /// # async fn run<I, S>(some_io: I, some_service: S)
455    /// # where
456    /// #     I: Read + Write + Unpin + Send + 'static,
457    /// #     S: Service<hyper::Request<Incoming>, Response=hyper::Response<Incoming>> + Send + 'static,
458    /// #     S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
459    /// #     S::Future: Send,
460    /// # {
461    /// let http = Builder::new();
462    /// let conn = http.serve_connection(some_io, some_service);
463    ///
464    /// if let Err(e) = conn.await {
465    ///     eprintln!("server connection error: {}", e);
466    /// }
467    /// # }
468    /// # fn main() {}
469    /// ```
470    pub fn serve_connection<I, S>(&self, io: I, service: S) -> Connection<I, S>
471    where
472        S: HttpService<IncomingBody>,
473        S::Error: Into<Box<dyn StdError + Send + Sync>>,
474        S::ResBody: 'static,
475        <S::ResBody as Body>::Error: Into<Box<dyn StdError + Send + Sync>>,
476        I: Read + Write + Unpin,
477    {
478        let mut conn = proto::Conn::new(io);
479        conn.set_h1_parser_config(self.h1_parser_config.clone());
480        conn.set_timer(self.timer.clone());
481        if !self.h1_keep_alive {
482            conn.disable_keep_alive();
483        }
484        if self.h1_half_close {
485            conn.set_allow_half_close();
486        }
487        if self.h1_title_case_headers {
488            conn.set_title_case_headers();
489        }
490        if self.h1_preserve_header_case {
491            conn.set_preserve_header_case();
492        }
493        if let Some(max_headers) = self.h1_max_headers {
494            conn.set_http1_max_headers(max_headers);
495        }
496        if let Some(max_header_size) = self.h1_max_header_size {
497            conn.set_http1_max_header_size(max_header_size);
498        }
499        if let Some(dur) = self
500            .timer
501            .check(self.h1_header_read_timeout, "header_read_timeout")
502        {
503            conn.set_http1_header_read_timeout(dur);
504        }
505        if let Some(writev) = self.h1_writev {
506            if writev {
507                conn.set_write_strategy_queue();
508            } else {
509                conn.set_write_strategy_flatten();
510            }
511        }
512        conn.set_flush_pipeline(self.pipeline_flush);
513        if let Some(max) = self.max_buf_size {
514            conn.set_max_buf_size(max);
515        }
516        if !self.date_header {
517            conn.disable_date_header();
518        }
519        let sd = proto::h1::dispatch::Server::new(service);
520        let proto = proto::h1::Dispatcher::new(sd, conn);
521        Connection { conn: proto }
522    }
523}
524
525/// A future binding a connection with a Service with Upgrade support.
526#[must_use = "futures do nothing unless polled"]
527#[allow(missing_debug_implementations)]
528pub struct UpgradeableConnection<T, S>
529where
530    S: HttpService<IncomingBody>,
531{
532    pub(super) inner: Option<Connection<T, S>>,
533}
534
535impl<I, B, S> UpgradeableConnection<I, S>
536where
537    S: HttpService<IncomingBody, ResBody = B>,
538    S::Error: Into<Box<dyn StdError + Send + Sync>>,
539    I: Read + Write + Unpin,
540    B: Body + 'static,
541    B::Error: Into<Box<dyn StdError + Send + Sync>>,
542{
543    /// Start a graceful shutdown process for this connection.
544    ///
545    /// This `Connection` should continue to be polled until shutdown
546    /// can finish.
547    pub fn graceful_shutdown(mut self: Pin<&mut Self>) {
548        // Connection (`inner`) is `None` if it was upgraded (and `poll` is `Ready`).
549        // In that case, we don't need to call `graceful_shutdown`.
550        if let Some(conn) = self.inner.as_mut() {
551            Pin::new(conn).graceful_shutdown();
552        }
553    }
554
555    /// Return the inner IO object, and additional information provided the connection
556    /// has not yet been upgraded.
557    pub fn into_parts(self) -> Option<Parts<I, S>> {
558        self.inner.map(|conn| conn.into_parts())
559    }
560}
561
562impl<I, B, S> Future for UpgradeableConnection<I, S>
563where
564    S: HttpService<IncomingBody, ResBody = B>,
565    S::Error: Into<Box<dyn StdError + Send + Sync>>,
566    I: Read + Write + Unpin + Send + 'static,
567    B: Body + 'static,
568    B::Error: Into<Box<dyn StdError + Send + Sync>>,
569{
570    type Output = crate::Result<()>;
571
572    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
573        if let Some(conn) = self.inner.as_mut() {
574            match ready!(Pin::new(&mut conn.conn).poll(cx)) {
575                Ok(proto::Dispatched::Shutdown) => Poll::Ready(Ok(())),
576                Ok(proto::Dispatched::Upgrade(pending)) => {
577                    let (io, buf, _) = self
578                        .inner
579                        .take()
580                        .expect("upgradeable server connection missing after upgrade")
581                        .conn
582                        .into_inner();
583                    pending.fulfill(Upgraded::new(io, buf));
584                    Poll::Ready(Ok(()))
585                }
586                Err(e) => Poll::Ready(Err(e)),
587            }
588        } else {
589            // inner is `None`, meaning the connection was upgraded, thus it's `Poll::Ready(Ok(()))`
590            Poll::Ready(Ok(()))
591        }
592    }
593}
hyper/server/conn/http1.rs

hyper/server/conn/
http1.rs
