hyper_util/server/conn/auto/
mod.rs

1//! Http1 or Http2 connection.
2
3pub mod upgrade;
4
5use hyper::service::HttpService;
6use std::marker::PhantomPinned;
7use std::mem::MaybeUninit;
8use std::pin::Pin;
9use std::task::{Context, Poll, ready};
10use std::{error::Error as StdError, io, time::Duration};
11
12use bytes::Bytes;
13use http::{Request, Response};
14use http_body::Body;
15use hyper::{
16    body::Incoming,
17    rt::{Read, ReadBuf, Timer, Write},
18    service::Service,
19};
20
21#[cfg(feature = "http1")]
22use hyper::server::conn::http1;
23
24#[cfg(feature = "http2")]
25use hyper::{rt::bounds::Http2ServerConnExec, server::conn::http2};
26
27#[cfg(any(not(feature = "http2"), not(feature = "http1")))]
28use std::marker::PhantomData;
29
30use pin_project_lite::pin_project;
31
32use crate::common::rewind::Rewind;
33
34type Error = Box<dyn std::error::Error + Send + Sync>;
35
36type Result<T> = std::result::Result<T, Error>;
37
38const H2_PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
39
40/// Exactly equivalent to [`Http2ServerConnExec`].
41#[cfg(feature = "http2")]
42pub trait HttpServerConnExec<A, B: Body>: Http2ServerConnExec<A, B> {}
43
44#[cfg(feature = "http2")]
45impl<A, B: Body, T: Http2ServerConnExec<A, B>> HttpServerConnExec<A, B> for T {}
46
47/// Exactly equivalent to [`Http2ServerConnExec`].
48#[cfg(not(feature = "http2"))]
49pub trait HttpServerConnExec<A, B: Body> {}
50
51#[cfg(not(feature = "http2"))]
52impl<A, B: Body, T> HttpServerConnExec<A, B> for T {}
53
54/// Http1 or Http2 connection builder.
55#[derive(Clone, Debug)]
56pub struct Builder<E> {
57    #[cfg(feature = "http1")]
58    http1: http1::Builder,
59    #[cfg(feature = "http2")]
60    http2: http2::Builder<E>,
61    #[cfg(any(feature = "http1", feature = "http2"))]
62    version: Option<Version>,
63    #[cfg(not(feature = "http2"))]
64    _executor: E,
65}
66
67impl<E: Default> Default for Builder<E> {
68    fn default() -> Self {
69        Self::new(E::default())
70    }
71}
72
73impl<E> Builder<E> {
74    /// Create a new auto connection builder.
75    ///
76    /// `executor` parameter should be a type that implements
77    /// [`Executor`](hyper::rt::Executor) trait.
78    ///
79    /// # Example
80    ///
81    /// ```
82    /// # #[cfg(feature = "tokio")]
83    /// # {
84    /// use hyper_util::{
85    ///     rt::TokioExecutor,
86    ///     server::conn::auto,
87    /// };
88    ///
89    /// auto::Builder::new(TokioExecutor::new());
90    /// # }
91    /// ```
92    pub fn new(executor: E) -> Self {
93        Self {
94            #[cfg(feature = "http1")]
95            http1: http1::Builder::new(),
96            #[cfg(feature = "http2")]
97            http2: http2::Builder::new(executor),
98            #[cfg(any(feature = "http1", feature = "http2"))]
99            version: None,
100            #[cfg(not(feature = "http2"))]
101            _executor: executor,
102        }
103    }
104
105    /// Http1 configuration.
106    #[cfg(feature = "http1")]
107    pub fn http1(&mut self) -> Http1Builder<'_, E> {
108        Http1Builder { inner: self }
109    }
110
111    /// Http2 configuration.
112    #[cfg(feature = "http2")]
113    pub fn http2(&mut self) -> Http2Builder<'_, E> {
114        Http2Builder { inner: self }
115    }
116
117    /// Only accepts HTTP/2
118    ///
119    /// Does not do anything if used with [`serve_connection_with_upgrades`]
120    ///
121    /// [`serve_connection_with_upgrades`]: Builder::serve_connection_with_upgrades
122    #[cfg(feature = "http2")]
123    pub fn http2_only(mut self) -> Self {
124        assert!(self.version.is_none());
125        self.version = Some(Version::H2);
126        self
127    }
128
129    /// Only accepts HTTP/1
130    ///
131    /// Does not do anything if used with [`serve_connection_with_upgrades`]
132    ///
133    /// [`serve_connection_with_upgrades`]: Builder::serve_connection_with_upgrades
134    #[cfg(feature = "http1")]
135    pub fn http1_only(mut self) -> Self {
136        assert!(self.version.is_none());
137        self.version = Some(Version::H1);
138        self
139    }
140
141    /// Returns `true` if this builder can serve an HTTP/1.1-based connection.
142    pub fn is_http1_available(&self) -> bool {
143        match self.version {
144            #[cfg(feature = "http1")]
145            Some(Version::H1) => true,
146            #[cfg(feature = "http2")]
147            Some(Version::H2) => false,
148            #[cfg(any(feature = "http1", feature = "http2"))]
149            _ => true,
150        }
151    }
152
153    /// Returns `true` if this builder can serve an HTTP/2-based connection.
154    pub fn is_http2_available(&self) -> bool {
155        match self.version {
156            #[cfg(feature = "http1")]
157            Some(Version::H1) => false,
158            #[cfg(feature = "http2")]
159            Some(Version::H2) => true,
160            #[cfg(any(feature = "http1", feature = "http2"))]
161            _ => true,
162        }
163    }
164
165    /// Set whether HTTP/1 connections will write header names as title case at
166    /// the socket level.
167    ///
168    /// This setting only affects HTTP/1 connections. HTTP/2 connections are
169    /// not affected by this setting.
170    ///
171    /// Default is false.
172    ///
173    /// # Example
174    ///
175    /// ```
176    /// # #[cfg(feature = "tokio")]
177    /// # {
178    /// use hyper_util::{
179    ///     rt::TokioExecutor,
180    ///     server::conn::auto,
181    /// };
182    ///
183    /// auto::Builder::new(TokioExecutor::new())
184    ///     .title_case_headers(true);
185    /// # }
186    /// ```
187    #[cfg(feature = "http1")]
188    pub fn title_case_headers(mut self, enabled: bool) -> Self {
189        self.http1.title_case_headers(enabled);
190        self
191    }
192
193    /// Set whether HTTP/1 connections will preserve the original case of header names.
194    ///
195    /// This setting only affects HTTP/1 connections. HTTP/2 connections are
196    /// not affected by this setting.
197    ///
198    /// Default is false.
199    ///
200    /// # Example
201    ///
202    /// ```
203    /// # #[cfg(feature = "tokio")]
204    /// # {
205    /// use hyper_util::{
206    ///     rt::TokioExecutor,
207    ///     server::conn::auto,
208    /// };
209    ///
210    /// auto::Builder::new(TokioExecutor::new())
211    ///     .preserve_header_case(true);
212    /// # }
213    /// ```
214    #[cfg(feature = "http1")]
215    pub fn preserve_header_case(mut self, enabled: bool) -> Self {
216        self.http1.preserve_header_case(enabled);
217        self
218    }
219
220    /// Bind a connection together with a [`Service`].
221    pub fn serve_connection<I, S, B>(&self, io: I, service: S) -> Connection<'_, I, S, E>
222    where
223        S: Service<Request<Incoming>, Response = Response<B>>,
224        S::Future: 'static,
225        S::Error: Into<Box<dyn StdError + Send + Sync>>,
226        B: Body + 'static,
227        B::Error: Into<Box<dyn StdError + Send + Sync>>,
228        I: Read + Write + Unpin + 'static,
229        E: HttpServerConnExec<S::Future, B>,
230    {
231        let state = match self.version {
232            #[cfg(feature = "http1")]
233            Some(Version::H1) => {
234                let io = Rewind::new_buffered(io, Bytes::new());
235                let conn = self.http1.serve_connection(io, service);
236                ConnState::H1 { conn }
237            }
238            #[cfg(feature = "http2")]
239            Some(Version::H2) => {
240                let io = Rewind::new_buffered(io, Bytes::new());
241                let conn = self.http2.serve_connection(io, service);
242                ConnState::H2 { conn }
243            }
244            #[cfg(any(feature = "http1", feature = "http2"))]
245            _ => ConnState::ReadVersion {
246                read_version: read_version(io),
247                builder: Cow::Borrowed(self),
248                service: Some(service),
249            },
250        };
251
252        Connection { state }
253    }
254
255    /// Bind a connection together with a [`Service`], with the ability to
256    /// handle HTTP upgrades. This requires that the IO object implements
257    /// `Send`.
258    ///
259    /// Note that if you ever want to use [`hyper::upgrade::Upgraded::downcast`]
260    /// with this crate, you'll need to use [`hyper_util::server::conn::auto::upgrade::downcast`]
261    /// instead. See the documentation of the latter to understand why.
262    ///
263    /// [`hyper_util::server::conn::auto::upgrade::downcast`]: crate::server::conn::auto::upgrade::downcast
264    pub fn serve_connection_with_upgrades<I, S, B>(
265        &self,
266        io: I,
267        service: S,
268    ) -> UpgradeableConnection<'_, I, S, E>
269    where
270        S: Service<Request<Incoming>, Response = Response<B>>,
271        S::Future: 'static,
272        S::Error: Into<Box<dyn StdError + Send + Sync>>,
273        B: Body + 'static,
274        B::Error: Into<Box<dyn StdError + Send + Sync>>,
275        I: Read + Write + Unpin + Send + 'static,
276        E: HttpServerConnExec<S::Future, B>,
277    {
278        UpgradeableConnection {
279            state: UpgradeableConnState::ReadVersion {
280                read_version: read_version(io),
281                builder: Cow::Borrowed(self),
282                service: Some(service),
283            },
284        }
285    }
286}
287
288#[derive(Copy, Clone, Debug)]
289enum Version {
290    H1,
291    H2,
292}
293
294impl Version {
295    #[must_use]
296    #[cfg(any(not(feature = "http2"), not(feature = "http1")))]
297    pub fn unsupported(self) -> Error {
298        match self {
299            Version::H1 => Error::from("HTTP/1 is not supported"),
300            Version::H2 => Error::from("HTTP/2 is not supported"),
301        }
302    }
303}
304
305fn read_version<I>(io: I) -> ReadVersion<I>
306where
307    I: Read + Unpin,
308{
309    ReadVersion {
310        io: Some(io),
311        buf: [MaybeUninit::uninit(); 24],
312        filled: 0,
313        version: Version::H2,
314        cancelled: false,
315        _pin: PhantomPinned,
316    }
317}
318
319pin_project! {
320    struct ReadVersion<I> {
321        io: Option<I>,
322        buf: [MaybeUninit<u8>; 24],
323        // the amount of `buf` thats been filled
324        filled: usize,
325        version: Version,
326        cancelled: bool,
327        // Make this future `!Unpin` for compatibility with async trait methods.
328        #[pin]
329        _pin: PhantomPinned,
330    }
331}
332
333impl<I> ReadVersion<I> {
334    pub fn cancel(self: Pin<&mut Self>) {
335        *self.project().cancelled = true;
336    }
337}
338
339impl<I> Future for ReadVersion<I>
340where
341    I: Read + Unpin,
342{
343    type Output = io::Result<(Version, Rewind<I>)>;
344
345    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
346        let this = self.project();
347        if *this.cancelled {
348            return Poll::Ready(Err(io::Error::new(io::ErrorKind::Interrupted, "Cancelled")));
349        }
350
351        let mut buf = ReadBuf::uninit(&mut *this.buf);
352        // SAFETY: `this.filled` tracks how many bytes have been read (and thus initialized) and
353        // we're only advancing by that many.
354        unsafe {
355            buf.unfilled().advance(*this.filled);
356        };
357
358        // We start as H2 and switch to H1 as soon as we don't have the preface.
359        while buf.filled().len() < H2_PREFACE.len() {
360            let len = buf.filled().len();
361            ready!(Pin::new(this.io.as_mut().unwrap()).poll_read(cx, buf.unfilled()))?;
362            *this.filled = buf.filled().len();
363
364            // We starts as H2 and switch to H1 when we don't get the preface.
365            if buf.filled().len() == len
366                || buf.filled()[len..] != H2_PREFACE[len..buf.filled().len()]
367            {
368                *this.version = Version::H1;
369                break;
370            }
371        }
372
373        let io = this.io.take().unwrap();
374        let buf = buf.filled().to_vec();
375        Poll::Ready(Ok((
376            *this.version,
377            Rewind::new_buffered(io, Bytes::from(buf)),
378        )))
379    }
380}
381
382pin_project! {
383    /// A [`Future`](core::future::Future) representing an HTTP/1 connection, returned from
384    /// [`Builder::serve_connection`](struct.Builder.html#method.serve_connection).
385    ///
386    /// To drive HTTP on this connection this future **must be polled**, typically with
387    /// `.await`. If it isn't polled, no progress will be made on this connection.
388    #[must_use = "futures do nothing unless polled"]
389    pub struct Connection<'a, I, S, E>
390    where
391        S: HttpService<Incoming>,
392    {
393        #[pin]
394        state: ConnState<'a, I, S, E>,
395    }
396}
397
398// A custom COW, since the libstd is has ToOwned bounds that are too eager.
399enum Cow<'a, T> {
400    Borrowed(&'a T),
401    Owned(T),
402}
403
404impl<T> std::ops::Deref for Cow<'_, T> {
405    type Target = T;
406    fn deref(&self) -> &T {
407        match self {
408            Cow::Borrowed(t) => &*t,
409            Cow::Owned(t) => t,
410        }
411    }
412}
413
414#[cfg(feature = "http1")]
415type Http1Connection<I, S> = hyper::server::conn::http1::Connection<Rewind<I>, S>;
416
417#[cfg(not(feature = "http1"))]
418type Http1Connection<I, S> = (PhantomData<I>, PhantomData<S>);
419
420#[cfg(feature = "http2")]
421type Http2Connection<I, S, E> = hyper::server::conn::http2::Connection<Rewind<I>, S, E>;
422
423#[cfg(not(feature = "http2"))]
424type Http2Connection<I, S, E> = (PhantomData<I>, PhantomData<S>, PhantomData<E>);
425
426pin_project! {
427    #[project = ConnStateProj]
428    enum ConnState<'a, I, S, E>
429    where
430        S: HttpService<Incoming>,
431    {
432        ReadVersion {
433            #[pin]
434            read_version: ReadVersion<I>,
435            builder: Cow<'a, Builder<E>>,
436            service: Option<S>,
437        },
438        H1 {
439            #[pin]
440            conn: Http1Connection<I, S>,
441        },
442        H2 {
443            #[pin]
444            conn: Http2Connection<I, S, E>,
445        },
446    }
447}
448
449impl<I, S, E, B> Connection<'_, I, S, E>
450where
451    S: HttpService<Incoming, ResBody = B>,
452    S::Error: Into<Box<dyn StdError + Send + Sync>>,
453    I: Read + Write + Unpin,
454    B: Body + 'static,
455    B::Error: Into<Box<dyn StdError + Send + Sync>>,
456    E: HttpServerConnExec<S::Future, B>,
457{
458    /// Start a graceful shutdown process for this connection.
459    ///
460    /// This `Connection` should continue to be polled until shutdown can finish.
461    ///
462    /// # Note
463    ///
464    /// This should only be called while the `Connection` future is still pending. If called after
465    /// `Connection::poll` has resolved, this does nothing.
466    pub fn graceful_shutdown(self: Pin<&mut Self>) {
467        match self.project().state.project() {
468            ConnStateProj::ReadVersion { read_version, .. } => read_version.cancel(),
469            #[cfg(feature = "http1")]
470            ConnStateProj::H1 { conn } => conn.graceful_shutdown(),
471            #[cfg(feature = "http2")]
472            ConnStateProj::H2 { conn } => conn.graceful_shutdown(),
473            #[cfg(any(not(feature = "http1"), not(feature = "http2")))]
474            _ => unreachable!(),
475        }
476    }
477
478    /// Make this Connection static, instead of borrowing from Builder.
479    pub fn into_owned(self) -> Connection<'static, I, S, E>
480    where
481        Builder<E>: Clone,
482    {
483        Connection {
484            state: match self.state {
485                ConnState::ReadVersion {
486                    read_version,
487                    builder,
488                    service,
489                } => ConnState::ReadVersion {
490                    read_version,
491                    service,
492                    builder: Cow::Owned(builder.clone()),
493                },
494                #[cfg(feature = "http1")]
495                ConnState::H1 { conn } => ConnState::H1 { conn },
496                #[cfg(feature = "http2")]
497                ConnState::H2 { conn } => ConnState::H2 { conn },
498                #[cfg(any(not(feature = "http1"), not(feature = "http2")))]
499                _ => unreachable!(),
500            },
501        }
502    }
503}
504
505impl<I, S, E, B> Future for Connection<'_, I, S, E>
506where
507    S: Service<Request<Incoming>, Response = Response<B>>,
508    S::Future: 'static,
509    S::Error: Into<Box<dyn StdError + Send + Sync>>,
510    B: Body + 'static,
511    B::Error: Into<Box<dyn StdError + Send + Sync>>,
512    I: Read + Write + Unpin + 'static,
513    E: HttpServerConnExec<S::Future, B>,
514{
515    type Output = Result<()>;
516
517    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
518        loop {
519            let mut this = self.as_mut().project();
520
521            match this.state.as_mut().project() {
522                ConnStateProj::ReadVersion {
523                    read_version,
524                    builder,
525                    service,
526                } => {
527                    let (version, io) = ready!(read_version.poll(cx))?;
528                    let service = service.take().unwrap();
529                    match version {
530                        #[cfg(feature = "http1")]
531                        Version::H1 => {
532                            let conn = builder.http1.serve_connection(io, service);
533                            this.state.set(ConnState::H1 { conn });
534                        }
535                        #[cfg(feature = "http2")]
536                        Version::H2 => {
537                            let conn = builder.http2.serve_connection(io, service);
538                            this.state.set(ConnState::H2 { conn });
539                        }
540                        #[cfg(any(not(feature = "http1"), not(feature = "http2")))]
541                        _ => return Poll::Ready(Err(version.unsupported())),
542                    }
543                }
544                #[cfg(feature = "http1")]
545                ConnStateProj::H1 { conn } => {
546                    return conn.poll(cx).map_err(Into::into);
547                }
548                #[cfg(feature = "http2")]
549                ConnStateProj::H2 { conn } => {
550                    return conn.poll(cx).map_err(Into::into);
551                }
552                #[cfg(any(not(feature = "http1"), not(feature = "http2")))]
553                _ => unreachable!(),
554            }
555        }
556    }
557}
558
559pin_project! {
560    /// An upgradable [`Connection`], returned by
561    /// [`Builder::serve_upgradable_connection`](struct.Builder.html#method.serve_connection_with_upgrades).
562    ///
563    /// To drive HTTP on this connection this future **must be polled**, typically with
564    /// `.await`. If it isn't polled, no progress will be made on this connection.
565    #[must_use = "futures do nothing unless polled"]
566    pub struct UpgradeableConnection<'a, I, S, E>
567    where
568        S: HttpService<Incoming>,
569    {
570        #[pin]
571        state: UpgradeableConnState<'a, I, S, E>,
572    }
573}
574
575#[cfg(feature = "http1")]
576type Http1UpgradeableConnection<I, S> = hyper::server::conn::http1::UpgradeableConnection<I, S>;
577
578#[cfg(not(feature = "http1"))]
579type Http1UpgradeableConnection<I, S> = (PhantomData<I>, PhantomData<S>);
580
581pin_project! {
582    #[project = UpgradeableConnStateProj]
583    enum UpgradeableConnState<'a, I, S, E>
584    where
585        S: HttpService<Incoming>,
586    {
587        ReadVersion {
588            #[pin]
589            read_version: ReadVersion<I>,
590            builder: Cow<'a, Builder<E>>,
591            service: Option<S>,
592        },
593        H1 {
594            #[pin]
595            conn: Http1UpgradeableConnection<Rewind<I>, S>,
596        },
597        H2 {
598            #[pin]
599            conn: Http2Connection<I, S, E>,
600        },
601    }
602}
603
604impl<I, S, E, B> UpgradeableConnection<'_, I, S, E>
605where
606    S: HttpService<Incoming, ResBody = B>,
607    S::Error: Into<Box<dyn StdError + Send + Sync>>,
608    I: Read + Write + Unpin,
609    B: Body + 'static,
610    B::Error: Into<Box<dyn StdError + Send + Sync>>,
611    E: HttpServerConnExec<S::Future, B>,
612{
613    /// Start a graceful shutdown process for this connection.
614    ///
615    /// This `UpgradeableConnection` should continue to be polled until shutdown can finish.
616    ///
617    /// # Note
618    ///
619    /// This should only be called while the `Connection` future is still nothing. pending. If
620    /// called after `UpgradeableConnection::poll` has resolved, this does nothing.
621    pub fn graceful_shutdown(self: Pin<&mut Self>) {
622        match self.project().state.project() {
623            UpgradeableConnStateProj::ReadVersion { read_version, .. } => read_version.cancel(),
624            #[cfg(feature = "http1")]
625            UpgradeableConnStateProj::H1 { conn } => conn.graceful_shutdown(),
626            #[cfg(feature = "http2")]
627            UpgradeableConnStateProj::H2 { conn } => conn.graceful_shutdown(),
628            #[cfg(any(not(feature = "http1"), not(feature = "http2")))]
629            _ => unreachable!(),
630        }
631    }
632
633    /// Make this Connection static, instead of borrowing from Builder.
634    pub fn into_owned(self) -> UpgradeableConnection<'static, I, S, E>
635    where
636        Builder<E>: Clone,
637    {
638        UpgradeableConnection {
639            state: match self.state {
640                UpgradeableConnState::ReadVersion {
641                    read_version,
642                    builder,
643                    service,
644                } => UpgradeableConnState::ReadVersion {
645                    read_version,
646                    service,
647                    builder: Cow::Owned(builder.clone()),
648                },
649                #[cfg(feature = "http1")]
650                UpgradeableConnState::H1 { conn } => UpgradeableConnState::H1 { conn },
651                #[cfg(feature = "http2")]
652                UpgradeableConnState::H2 { conn } => UpgradeableConnState::H2 { conn },
653                #[cfg(any(not(feature = "http1"), not(feature = "http2")))]
654                _ => unreachable!(),
655            },
656        }
657    }
658}
659
660impl<I, S, E, B> Future for UpgradeableConnection<'_, I, S, E>
661where
662    S: Service<Request<Incoming>, Response = Response<B>>,
663    S::Future: 'static,
664    S::Error: Into<Box<dyn StdError + Send + Sync>>,
665    B: Body + 'static,
666    B::Error: Into<Box<dyn StdError + Send + Sync>>,
667    I: Read + Write + Unpin + Send + 'static,
668    E: HttpServerConnExec<S::Future, B>,
669{
670    type Output = Result<()>;
671
672    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
673        loop {
674            let mut this = self.as_mut().project();
675
676            match this.state.as_mut().project() {
677                UpgradeableConnStateProj::ReadVersion {
678                    read_version,
679                    builder,
680                    service,
681                } => {
682                    let (version, io) = ready!(read_version.poll(cx))?;
683                    let service = service.take().unwrap();
684                    match version {
685                        #[cfg(feature = "http1")]
686                        Version::H1 => {
687                            let conn = builder.http1.serve_connection(io, service).with_upgrades();
688                            this.state.set(UpgradeableConnState::H1 { conn });
689                        }
690                        #[cfg(feature = "http2")]
691                        Version::H2 => {
692                            let conn = builder.http2.serve_connection(io, service);
693                            this.state.set(UpgradeableConnState::H2 { conn });
694                        }
695                        #[cfg(any(not(feature = "http1"), not(feature = "http2")))]
696                        _ => return Poll::Ready(Err(version.unsupported())),
697                    }
698                }
699                #[cfg(feature = "http1")]
700                UpgradeableConnStateProj::H1 { conn } => {
701                    return conn.poll(cx).map_err(Into::into);
702                }
703                #[cfg(feature = "http2")]
704                UpgradeableConnStateProj::H2 { conn } => {
705                    return conn.poll(cx).map_err(Into::into);
706                }
707                #[cfg(any(not(feature = "http1"), not(feature = "http2")))]
708                _ => unreachable!(),
709            }
710        }
711    }
712}
713
714/// Http1 part of builder.
715#[cfg(feature = "http1")]
716pub struct Http1Builder<'a, E> {
717    inner: &'a mut Builder<E>,
718}
719
720#[cfg(feature = "http1")]
721impl<E> Http1Builder<'_, E> {
722    /// Http2 configuration.
723    #[cfg(feature = "http2")]
724    pub fn http2(&mut self) -> Http2Builder<'_, E> {
725        Http2Builder { inner: self.inner }
726    }
727
728    /// Set whether the `date` header should be included in HTTP responses.
729    ///
730    /// Note that including the `date` header is recommended by RFC 7231.
731    ///
732    /// Default is true.
733    pub fn auto_date_header(&mut self, enabled: bool) -> &mut Self {
734        self.inner.http1.auto_date_header(enabled);
735        self
736    }
737
738    /// Set whether HTTP/1 connections should support half-closures.
739    ///
740    /// Clients can chose to shutdown their write-side while waiting
741    /// for the server to respond. Setting this to `true` will
742    /// prevent closing the connection immediately if `read`
743    /// detects an EOF in the middle of a request.
744    ///
745    /// Default is `false`.
746    pub fn half_close(&mut self, val: bool) -> &mut Self {
747        self.inner.http1.half_close(val);
748        self
749    }
750
751    /// Enables or disables HTTP/1 keep-alive.
752    ///
753    /// Default is true.
754    pub fn keep_alive(&mut self, val: bool) -> &mut Self {
755        self.inner.http1.keep_alive(val);
756        self
757    }
758
759    /// Set whether HTTP/1 connections will write header names as title case at
760    /// the socket level.
761    ///
762    /// Note that this setting does not affect HTTP/2.
763    ///
764    /// Default is false.
765    pub fn title_case_headers(&mut self, enabled: bool) -> &mut Self {
766        self.inner.http1.title_case_headers(enabled);
767        self
768    }
769
770    /// Set whether HTTP/1 connections will silently ignored malformed header lines.
771    ///
772    /// If this is enabled and a header line does not start with a valid header
773    /// name, or does not include a colon at all, the line will be silently ignored
774    /// and no error will be reported.
775    ///
776    /// Default is false.
777    pub fn ignore_invalid_headers(&mut self, enabled: bool) -> &mut Self {
778        self.inner.http1.ignore_invalid_headers(enabled);
779        self
780    }
781
782    /// Set whether to support preserving original header cases.
783    ///
784    /// Currently, this will record the original cases received, and store them
785    /// in a private extension on the `Request`. It will also look for and use
786    /// such an extension in any provided `Response`.
787    ///
788    /// Since the relevant extension is still private, there is no way to
789    /// interact with the original cases. The only effect this can have now is
790    /// to forward the cases in a proxy-like fashion.
791    ///
792    /// Note that this setting does not affect HTTP/2.
793    ///
794    /// Default is false.
795    pub fn preserve_header_case(&mut self, enabled: bool) -> &mut Self {
796        self.inner.http1.preserve_header_case(enabled);
797        self
798    }
799
800    /// Set the maximum number of headers.
801    ///
802    /// When a request is received, the parser will reserve a buffer to store headers for optimal
803    /// performance.
804    ///
805    /// If server receives more headers than the buffer size, it responds to the client with
806    /// "431 Request Header Fields Too Large".
807    ///
808    /// The headers is allocated on the stack by default, which has higher performance. After
809    /// setting this value, headers will be allocated in heap memory, that is, heap memory
810    /// allocation will occur for each request, and there will be a performance drop of about 5%.
811    ///
812    /// Note that this setting does not affect HTTP/2.
813    ///
814    /// Default is 100.
815    pub fn max_headers(&mut self, val: usize) -> &mut Self {
816        self.inner.http1.max_headers(val);
817        self
818    }
819
820    /// Set a timeout for reading client request headers. If a client does not
821    /// transmit the entire header within this time, the connection is closed.
822    ///
823    /// Requires a [`Timer`] set by [`Http1Builder::timer`] to take effect. Panics if `header_read_timeout` is configured
824    /// without a [`Timer`].
825    ///
826    /// Pass `None` to disable.
827    ///
828    /// Default is currently 30 seconds, but do not depend on that.
829    pub fn header_read_timeout(&mut self, read_timeout: impl Into<Option<Duration>>) -> &mut Self {
830        self.inner.http1.header_read_timeout(read_timeout);
831        self
832    }
833
834    /// Set whether HTTP/1 connections should try to use vectored writes,
835    /// or always flatten into a single buffer.
836    ///
837    /// Note that setting this to false may mean more copies of body data,
838    /// but may also improve performance when an IO transport doesn't
839    /// support vectored writes well, such as most TLS implementations.
840    ///
841    /// Setting this to true will force hyper to use queued strategy
842    /// which may eliminate unnecessary cloning on some TLS backends
843    ///
844    /// Default is `auto`. In this mode hyper will try to guess which
845    /// mode to use
846    pub fn writev(&mut self, val: bool) -> &mut Self {
847        self.inner.http1.writev(val);
848        self
849    }
850
851    /// Set the maximum buffer size for the connection.
852    ///
853    /// Default is ~400kb.
854    ///
855    /// # Panics
856    ///
857    /// The minimum value allowed is 8192. This method panics if the passed `max` is less than the minimum.
858    pub fn max_buf_size(&mut self, max: usize) -> &mut Self {
859        self.inner.http1.max_buf_size(max);
860        self
861    }
862
863    /// Aggregates flushes to better support pipelined responses.
864    ///
865    /// Experimental, may have bugs.
866    ///
867    /// Default is false.
868    pub fn pipeline_flush(&mut self, enabled: bool) -> &mut Self {
869        self.inner.http1.pipeline_flush(enabled);
870        self
871    }
872
873    /// Set the timer used in background tasks.
874    pub fn timer<M>(&mut self, timer: M) -> &mut Self
875    where
876        M: Timer + Send + Sync + 'static,
877    {
878        self.inner.http1.timer(timer);
879        self
880    }
881
882    /// Bind a connection together with a [`Service`].
883    #[cfg(feature = "http2")]
884    pub async fn serve_connection<I, S, B>(&self, io: I, service: S) -> Result<()>
885    where
886        S: Service<Request<Incoming>, Response = Response<B>>,
887        S::Future: 'static,
888        S::Error: Into<Box<dyn StdError + Send + Sync>>,
889        B: Body + 'static,
890        B::Error: Into<Box<dyn StdError + Send + Sync>>,
891        I: Read + Write + Unpin + 'static,
892        E: HttpServerConnExec<S::Future, B>,
893    {
894        self.inner.serve_connection(io, service).await
895    }
896
897    /// Bind a connection together with a [`Service`].
898    #[cfg(not(feature = "http2"))]
899    pub async fn serve_connection<I, S, B>(&self, io: I, service: S) -> Result<()>
900    where
901        S: Service<Request<Incoming>, Response = Response<B>>,
902        S::Future: 'static,
903        S::Error: Into<Box<dyn StdError + Send + Sync>>,
904        B: Body + 'static,
905        B::Error: Into<Box<dyn StdError + Send + Sync>>,
906        I: Read + Write + Unpin + 'static,
907    {
908        self.inner.serve_connection(io, service).await
909    }
910
911    /// Bind a connection together with a [`Service`], with the ability to
912    /// handle HTTP upgrades. This requires that the IO object implements
913    /// `Send`.
914    #[cfg(feature = "http2")]
915    pub fn serve_connection_with_upgrades<I, S, B>(
916        &self,
917        io: I,
918        service: S,
919    ) -> UpgradeableConnection<'_, I, S, E>
920    where
921        S: Service<Request<Incoming>, Response = Response<B>>,
922        S::Future: 'static,
923        S::Error: Into<Box<dyn StdError + Send + Sync>>,
924        B: Body + 'static,
925        B::Error: Into<Box<dyn StdError + Send + Sync>>,
926        I: Read + Write + Unpin + Send + 'static,
927        E: HttpServerConnExec<S::Future, B>,
928    {
929        self.inner.serve_connection_with_upgrades(io, service)
930    }
931}
932
933/// Http2 part of builder.
934#[cfg(feature = "http2")]
935pub struct Http2Builder<'a, E> {
936    inner: &'a mut Builder<E>,
937}
938
939#[cfg(feature = "http2")]
940impl<E> Http2Builder<'_, E> {
941    #[cfg(feature = "http1")]
942    /// Http1 configuration.
943    pub fn http1(&mut self) -> Http1Builder<'_, E> {
944        Http1Builder { inner: self.inner }
945    }
946
947    /// Configures the maximum number of pending reset streams allowed before a GOAWAY will be sent.
948    ///
949    /// This will default to the default value set by the [`h2` crate](https://crates.io/crates/h2).
950    /// As of v0.4.0, it is 20.
951    ///
952    /// See <https://github.com/hyperium/hyper/issues/2877> for more information.
953    pub fn max_pending_accept_reset_streams(&mut self, max: impl Into<Option<usize>>) -> &mut Self {
954        self.inner.http2.max_pending_accept_reset_streams(max);
955        self
956    }
957
958    /// Configures the maximum number of local reset streams allowed before a GOAWAY will be sent.
959    ///
960    /// If not set, hyper will use a default, currently of 1024.
961    ///
962    /// If `None` is supplied, hyper will not apply any limit.
963    /// This is not advised, as it can potentially expose servers to DOS vulnerabilities.
964    ///
965    /// See <https://rustsec.org/advisories/RUSTSEC-2024-0003.html> for more information.
966    pub fn max_local_error_reset_streams(&mut self, max: impl Into<Option<usize>>) -> &mut Self {
967        self.inner.http2.max_local_error_reset_streams(max);
968        self
969    }
970
971    /// Sets the [`SETTINGS_INITIAL_WINDOW_SIZE`][spec] option for HTTP2
972    /// stream-level flow control.
973    ///
974    /// Passing `None` will do nothing.
975    ///
976    /// If not set, hyper will use a default.
977    ///
978    /// [spec]: https://http2.github.io/http2-spec/#SETTINGS_INITIAL_WINDOW_SIZE
979    pub fn initial_stream_window_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
980        self.inner.http2.initial_stream_window_size(sz);
981        self
982    }
983
984    /// Sets the max connection-level flow control for HTTP2.
985    ///
986    /// Passing `None` will do nothing.
987    ///
988    /// If not set, hyper will use a default.
989    pub fn initial_connection_window_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
990        self.inner.http2.initial_connection_window_size(sz);
991        self
992    }
993
994    /// Sets whether to use an adaptive flow control.
995    ///
996    /// Enabling this will override the limits set in
997    /// `http2_initial_stream_window_size` and
998    /// `http2_initial_connection_window_size`.
999    pub fn adaptive_window(&mut self, enabled: bool) -> &mut Self {
1000        self.inner.http2.adaptive_window(enabled);
1001        self
1002    }
1003
1004    /// Sets the maximum frame size to use for HTTP2.
1005    ///
1006    /// Passing `None` will do nothing.
1007    ///
1008    /// If not set, hyper will use a default.
1009    pub fn max_frame_size(&mut self, sz: impl Into<Option<u32>>) -> &mut Self {
1010        self.inner.http2.max_frame_size(sz);
1011        self
1012    }
1013
1014    /// Sets the [`SETTINGS_MAX_CONCURRENT_STREAMS`][spec] option for HTTP2
1015    /// connections.
1016    ///
1017    /// Default is 200. Passing `None` will remove any limit.
1018    ///
1019    /// [spec]: https://http2.github.io/http2-spec/#SETTINGS_MAX_CONCURRENT_STREAMS
1020    pub fn max_concurrent_streams(&mut self, max: impl Into<Option<u32>>) -> &mut Self {
1021        self.inner.http2.max_concurrent_streams(max);
1022        self
1023    }
1024
1025    /// Sets an interval for HTTP2 Ping frames should be sent to keep a
1026    /// connection alive.
1027    ///
1028    /// Pass `None` to disable HTTP2 keep-alive.
1029    ///
1030    /// Default is currently disabled.
1031    ///
1032    /// # Cargo Feature
1033    ///
1034    pub fn keep_alive_interval(&mut self, interval: impl Into<Option<Duration>>) -> &mut Self {
1035        self.inner.http2.keep_alive_interval(interval);
1036        self
1037    }
1038
1039    /// Sets a timeout for receiving an acknowledgement of the keep-alive ping.
1040    ///
1041    /// If the ping is not acknowledged within the timeout, the connection will
1042    /// be closed. Does nothing if `http2_keep_alive_interval` is disabled.
1043    ///
1044    /// Default is 20 seconds.
1045    ///
1046    /// # Cargo Feature
1047    ///
1048    pub fn keep_alive_timeout(&mut self, timeout: Duration) -> &mut Self {
1049        self.inner.http2.keep_alive_timeout(timeout);
1050        self
1051    }
1052
1053    /// Set the maximum write buffer size for each HTTP/2 stream.
1054    ///
1055    /// Default is currently ~400KB, but may change.
1056    ///
1057    /// # Panics
1058    ///
1059    /// The value must be no larger than `u32::MAX`.
1060    pub fn max_send_buf_size(&mut self, max: usize) -> &mut Self {
1061        self.inner.http2.max_send_buf_size(max);
1062        self
1063    }
1064
1065    /// Enables the [extended CONNECT protocol].
1066    ///
1067    /// [extended CONNECT protocol]: https://datatracker.ietf.org/doc/html/rfc8441#section-4
1068    pub fn enable_connect_protocol(&mut self) -> &mut Self {
1069        self.inner.http2.enable_connect_protocol();
1070        self
1071    }
1072
1073    /// Sets the max size of received header frames.
1074    ///
1075    /// Default is currently ~16MB, but may change.
1076    pub fn max_header_list_size(&mut self, max: u32) -> &mut Self {
1077        self.inner.http2.max_header_list_size(max);
1078        self
1079    }
1080
1081    /// Set the timer used in background tasks.
1082    pub fn timer<M>(&mut self, timer: M) -> &mut Self
1083    where
1084        M: Timer + Send + Sync + 'static,
1085    {
1086        self.inner.http2.timer(timer);
1087        self
1088    }
1089
1090    /// Set whether the `date` header should be included in HTTP responses.
1091    ///
1092    /// Note that including the `date` header is recommended by RFC 7231.
1093    ///
1094    /// Default is true.
1095    pub fn auto_date_header(&mut self, enabled: bool) -> &mut Self {
1096        self.inner.http2.auto_date_header(enabled);
1097        self
1098    }
1099
1100    /// Bind a connection together with a [`Service`].
1101    pub async fn serve_connection<I, S, B>(&self, io: I, service: S) -> Result<()>
1102    where
1103        S: Service<Request<Incoming>, Response = Response<B>>,
1104        S::Future: 'static,
1105        S::Error: Into<Box<dyn StdError + Send + Sync>>,
1106        B: Body + 'static,
1107        B::Error: Into<Box<dyn StdError + Send + Sync>>,
1108        I: Read + Write + Unpin + 'static,
1109        E: HttpServerConnExec<S::Future, B>,
1110    {
1111        self.inner.serve_connection(io, service).await
1112    }
1113
1114    /// Bind a connection together with a [`Service`], with the ability to
1115    /// handle HTTP upgrades. This requires that the IO object implements
1116    /// `Send`.
1117    pub fn serve_connection_with_upgrades<I, S, B>(
1118        &self,
1119        io: I,
1120        service: S,
1121    ) -> UpgradeableConnection<'_, I, S, E>
1122    where
1123        S: Service<Request<Incoming>, Response = Response<B>>,
1124        S::Future: 'static,
1125        S::Error: Into<Box<dyn StdError + Send + Sync>>,
1126        B: Body + 'static,
1127        B::Error: Into<Box<dyn StdError + Send + Sync>>,
1128        I: Read + Write + Unpin + Send + 'static,
1129        E: HttpServerConnExec<S::Future, B>,
1130    {
1131        self.inner.serve_connection_with_upgrades(io, service)
1132    }
1133}
1134
1135#[cfg(all(feature = "tokio", test))]
1136mod tests {
1137    use crate::{
1138        rt::{TokioExecutor, TokioIo},
1139        server::conn::auto,
1140    };
1141    use http::{Request, Response};
1142    use http_body::Body;
1143    use http_body_util::{BodyExt, Empty, Full};
1144    use hyper::{body, body::Bytes, client, service::service_fn};
1145    use std::{convert::Infallible, error::Error as StdError, net::SocketAddr, time::Duration};
1146    use tokio::{
1147        net::{TcpListener, TcpStream},
1148        pin,
1149    };
1150
1151    const BODY: &[u8] = b"Hello, world!";
1152
1153    #[test]
1154    fn configuration() {
1155        // One liner.
1156        auto::Builder::new(TokioExecutor::new())
1157            .http1()
1158            .keep_alive(true)
1159            .http2()
1160            .keep_alive_interval(None);
1161        //  .serve_connection(io, service);
1162
1163        // Using variable.
1164        let mut builder = auto::Builder::new(TokioExecutor::new());
1165
1166        builder.http1().keep_alive(true);
1167        builder.http2().keep_alive_interval(None);
1168        // builder.serve_connection(io, service);
1169    }
1170
1171    #[test]
1172    #[cfg(feature = "http1")]
1173    fn title_case_headers_configuration() {
1174        // Test title_case_headers can be set on the main builder
1175        auto::Builder::new(TokioExecutor::new()).title_case_headers(true);
1176
1177        // Can be combined with other configuration
1178        auto::Builder::new(TokioExecutor::new())
1179            .title_case_headers(true)
1180            .http1_only();
1181    }
1182
1183    #[test]
1184    #[cfg(feature = "http1")]
1185    fn preserve_header_case_configuration() {
1186        // Test preserve_header_case can be set on the main builder
1187        auto::Builder::new(TokioExecutor::new()).preserve_header_case(true);
1188
1189        // Can be combined with other configuration
1190        auto::Builder::new(TokioExecutor::new())
1191            .preserve_header_case(true)
1192            .http1_only();
1193    }
1194
1195    #[cfg(not(miri))]
1196    #[tokio::test]
1197    async fn http1() {
1198        let addr = start_server(false, false).await;
1199        let mut sender = connect_h1(addr).await;
1200
1201        let response = sender
1202            .send_request(Request::new(Empty::<Bytes>::new()))
1203            .await
1204            .unwrap();
1205
1206        let body = response.into_body().collect().await.unwrap().to_bytes();
1207
1208        assert_eq!(body, BODY);
1209    }
1210
1211    #[cfg(not(miri))]
1212    #[tokio::test]
1213    async fn http2() {
1214        let addr = start_server(false, false).await;
1215        let mut sender = connect_h2(addr).await;
1216
1217        let response = sender
1218            .send_request(Request::new(Empty::<Bytes>::new()))
1219            .await
1220            .unwrap();
1221
1222        let body = response.into_body().collect().await.unwrap().to_bytes();
1223
1224        assert_eq!(body, BODY);
1225    }
1226
1227    #[cfg(not(miri))]
1228    #[tokio::test]
1229    async fn http2_only() {
1230        let addr = start_server(false, true).await;
1231        let mut sender = connect_h2(addr).await;
1232
1233        let response = sender
1234            .send_request(Request::new(Empty::<Bytes>::new()))
1235            .await
1236            .unwrap();
1237
1238        let body = response.into_body().collect().await.unwrap().to_bytes();
1239
1240        assert_eq!(body, BODY);
1241    }
1242
1243    #[cfg(not(miri))]
1244    #[tokio::test]
1245    async fn http2_only_fail_if_client_is_http1() {
1246        let addr = start_server(false, true).await;
1247        let mut sender = connect_h1(addr).await;
1248
1249        let _ = sender
1250            .send_request(Request::new(Empty::<Bytes>::new()))
1251            .await
1252            .expect_err("should fail");
1253    }
1254
1255    #[cfg(not(miri))]
1256    #[tokio::test]
1257    async fn http1_only() {
1258        let addr = start_server(true, false).await;
1259        let mut sender = connect_h1(addr).await;
1260
1261        let response = sender
1262            .send_request(Request::new(Empty::<Bytes>::new()))
1263            .await
1264            .unwrap();
1265
1266        let body = response.into_body().collect().await.unwrap().to_bytes();
1267
1268        assert_eq!(body, BODY);
1269    }
1270
1271    #[cfg(not(miri))]
1272    #[tokio::test]
1273    async fn http1_only_fail_if_client_is_http2() {
1274        let addr = start_server(true, false).await;
1275        let mut sender = connect_h2(addr).await;
1276
1277        let _ = sender
1278            .send_request(Request::new(Empty::<Bytes>::new()))
1279            .await
1280            .expect_err("should fail");
1281    }
1282
1283    #[cfg(not(miri))]
1284    #[tokio::test]
1285    async fn graceful_shutdown() {
1286        let listener = TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0)))
1287            .await
1288            .unwrap();
1289
1290        let listener_addr = listener.local_addr().unwrap();
1291
1292        // Spawn the task in background so that we can connect there
1293        let listen_task = tokio::spawn(async move { listener.accept().await.unwrap() });
1294        // Only connect a stream, do not send headers or anything
1295        let _stream = TcpStream::connect(listener_addr).await.unwrap();
1296
1297        let (stream, _) = listen_task.await.unwrap();
1298        let stream = TokioIo::new(stream);
1299        let builder = auto::Builder::new(TokioExecutor::new());
1300        let connection = builder.serve_connection(stream, service_fn(hello));
1301
1302        pin!(connection);
1303
1304        connection.as_mut().graceful_shutdown();
1305
1306        let connection_error = tokio::time::timeout(Duration::from_millis(200), connection)
1307            .await
1308            .expect("Connection should have finished in a timely manner after graceful shutdown.")
1309            .expect_err("Connection should have been interrupted.");
1310
1311        let connection_error = connection_error
1312            .downcast_ref::<std::io::Error>()
1313            .expect("The error should have been `std::io::Error`.");
1314        assert_eq!(connection_error.kind(), std::io::ErrorKind::Interrupted);
1315    }
1316
1317    async fn connect_h1<B>(addr: SocketAddr) -> client::conn::http1::SendRequest<B>
1318    where
1319        B: Body + Send + 'static,
1320        B::Data: Send,
1321        B::Error: Into<Box<dyn StdError + Send + Sync>>,
1322    {
1323        let stream = TokioIo::new(TcpStream::connect(addr).await.unwrap());
1324        let (sender, connection) = client::conn::http1::handshake(stream).await.unwrap();
1325
1326        tokio::spawn(connection);
1327
1328        sender
1329    }
1330
1331    async fn connect_h2<B>(addr: SocketAddr) -> client::conn::http2::SendRequest<B>
1332    where
1333        B: Body + Unpin + Send + 'static,
1334        B::Data: Send,
1335        B::Error: Into<Box<dyn StdError + Send + Sync>>,
1336    {
1337        let stream = TokioIo::new(TcpStream::connect(addr).await.unwrap());
1338        let (sender, connection) = client::conn::http2::Builder::new(TokioExecutor::new())
1339            .handshake(stream)
1340            .await
1341            .unwrap();
1342
1343        tokio::spawn(connection);
1344
1345        sender
1346    }
1347
1348    async fn start_server(h1_only: bool, h2_only: bool) -> SocketAddr {
1349        let addr: SocketAddr = ([127, 0, 0, 1], 0).into();
1350        let listener = TcpListener::bind(addr).await.unwrap();
1351
1352        let local_addr = listener.local_addr().unwrap();
1353
1354        tokio::spawn(async move {
1355            loop {
1356                let (stream, _) = listener.accept().await.unwrap();
1357                let stream = TokioIo::new(stream);
1358                tokio::task::spawn(async move {
1359                    let mut builder = auto::Builder::new(TokioExecutor::new());
1360                    if h1_only {
1361                        builder = builder.http1_only();
1362                        builder.serve_connection(stream, service_fn(hello)).await
1363                    } else if h2_only {
1364                        builder = builder.http2_only();
1365                        builder.serve_connection(stream, service_fn(hello)).await
1366                    } else {
1367                        builder
1368                            .http2()
1369                            .max_header_list_size(4096)
1370                            .serve_connection_with_upgrades(stream, service_fn(hello))
1371                            .await
1372                    }
1373                    .unwrap();
1374                });
1375            }
1376        });
1377
1378        local_addr
1379    }
1380
1381    async fn hello(_req: Request<body::Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
1382        Ok(Response::new(Full::new(Bytes::from(BODY))))
1383    }
1384}
hyper_util/server/conn/auto/mod.rs

hyper_util/server/conn/auto/
mod.rs
