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

hyper/server/conn/
http1.rs
