hyper/server/
server.rs

1use std::error::Error as StdError;
2use std::fmt;
3#[cfg(feature = "tcp")]
4use std::net::{SocketAddr, TcpListener as StdTcpListener};
5#[cfg(any(feature = "tcp", feature = "http1"))]
6use std::time::Duration;
7
8use pin_project_lite::pin_project;
9use tokio::io::{AsyncRead, AsyncWrite};
10use tracing::trace;
11
12use super::accept::Accept;
13#[cfg(all(feature = "tcp"))]
14use super::tcp::AddrIncoming;
15use crate::body::{Body, HttpBody};
16use crate::common::exec::Exec;
17use crate::common::exec::{ConnStreamExec, NewSvcExec};
18use crate::common::{task, Future, Pin, Poll, Unpin};
19// Renamed `Http` as `Http_` for now so that people upgrading don't see an
20// error that `hyper::server::Http` is private...
21use super::conn::{Connection, Http as Http_, UpgradeableConnection};
22use super::shutdown::{Graceful, GracefulWatcher};
23use crate::service::{HttpService, MakeServiceRef};
24
25use self::new_svc::NewSvcTask;
26
27pin_project! {
28    /// A listening HTTP server that accepts connections in both HTTP1 and HTTP2 by default.
29    ///
30    /// `Server` is a `Future` mapping a bound listener with a set of service
31    /// handlers. It is built using the [`Builder`](Builder), and the future
32    /// completes when the server has been shutdown. It should be run by an
33    /// `Executor`.
34    pub struct Server<I, S, E = Exec> {
35        #[pin]
36        incoming: I,
37        make_service: S,
38        protocol: Http_<E>,
39    }
40}
41
42/// A builder for a [`Server`](Server).
43#[derive(Debug)]
44#[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
45pub struct Builder<I, E = Exec> {
46    incoming: I,
47    protocol: Http_<E>,
48}
49
50// ===== impl Server =====
51
52#[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
53impl<I> Server<I, ()> {
54    /// Starts a [`Builder`](Builder) with the provided incoming stream.
55    pub fn builder(incoming: I) -> Builder<I> {
56        Builder {
57            incoming,
58            protocol: Http_::new(),
59        }
60    }
61}
62
63#[cfg(feature = "tcp")]
64#[cfg_attr(
65    docsrs,
66    doc(cfg(all(feature = "tcp", any(feature = "http1", feature = "http2"))))
67)]
68impl Server<AddrIncoming, ()> {
69    /// Binds to the provided address, and returns a [`Builder`](Builder).
70    ///
71    /// # Panics
72    ///
73    /// This method will panic if binding to the address fails. For a method
74    /// to bind to an address and return a `Result`, see `Server::try_bind`.
75    pub fn bind(addr: &SocketAddr) -> Builder<AddrIncoming> {
76        let incoming = AddrIncoming::new(addr).unwrap_or_else(|e| {
77            panic!("error binding to {}: {}", addr, e);
78        });
79        Server::builder(incoming)
80    }
81
82    /// Tries to bind to the provided address, and returns a [`Builder`](Builder).
83    pub fn try_bind(addr: &SocketAddr) -> crate::Result<Builder<AddrIncoming>> {
84        AddrIncoming::new(addr).map(Server::builder)
85    }
86
87    /// Create a new instance from a `std::net::TcpListener` instance.
88    pub fn from_tcp(listener: StdTcpListener) -> Result<Builder<AddrIncoming>, crate::Error> {
89        AddrIncoming::from_std(listener).map(Server::builder)
90    }
91}
92
93#[cfg(feature = "tcp")]
94#[cfg_attr(
95    docsrs,
96    doc(cfg(all(feature = "tcp", any(feature = "http1", feature = "http2"))))
97)]
98impl<S, E> Server<AddrIncoming, S, E> {
99    /// Returns the local address that this server is bound to.
100    pub fn local_addr(&self) -> SocketAddr {
101        self.incoming.local_addr()
102    }
103}
104
105#[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
106impl<I, IO, IE, S, E, B> Server<I, S, E>
107where
108    I: Accept<Conn = IO, Error = IE>,
109    IE: Into<Box<dyn StdError + Send + Sync>>,
110    IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
111    S: MakeServiceRef<IO, Body, ResBody = B>,
112    S::Error: Into<Box<dyn StdError + Send + Sync>>,
113    B: HttpBody + 'static,
114    B::Error: Into<Box<dyn StdError + Send + Sync>>,
115    E: ConnStreamExec<<S::Service as HttpService<Body>>::Future, B>,
116{
117    /// Prepares a server to handle graceful shutdown when the provided future
118    /// completes.
119    ///
120    /// # Example
121    ///
122    /// ```
123    /// # fn main() {}
124    /// # #[cfg(feature = "tcp")]
125    /// # async fn run() {
126    /// # use hyper::{Body, Response, Server, Error};
127    /// # use hyper::service::{make_service_fn, service_fn};
128    /// # let make_service = make_service_fn(|_| async {
129    /// #     Ok::<_, Error>(service_fn(|_req| async {
130    /// #         Ok::<_, Error>(Response::new(Body::from("Hello World")))
131    /// #     }))
132    /// # });
133    /// // Make a server from the previous examples...
134    /// let server = Server::bind(&([127, 0, 0, 1], 3000).into())
135    ///     .serve(make_service);
136    ///
137    /// // Prepare some signal for when the server should start shutting down...
138    /// let (tx, rx) = tokio::sync::oneshot::channel::<()>();
139    /// let graceful = server
140    ///     .with_graceful_shutdown(async {
141    ///         rx.await.ok();
142    ///     });
143    ///
144    /// // Await the `server` receiving the signal...
145    /// if let Err(e) = graceful.await {
146    ///     eprintln!("server error: {}", e);
147    /// }
148    ///
149    /// // And later, trigger the signal by calling `tx.send(())`.
150    /// let _ = tx.send(());
151    /// # }
152    /// ```
153    pub fn with_graceful_shutdown<F>(self, signal: F) -> Graceful<I, S, F, E>
154    where
155        F: Future<Output = ()>,
156        E: NewSvcExec<IO, S::Future, S::Service, E, GracefulWatcher>,
157    {
158        Graceful::new(self, signal)
159    }
160
161    fn poll_next_(
162        self: Pin<&mut Self>,
163        cx: &mut task::Context<'_>,
164    ) -> Poll<Option<crate::Result<Connecting<IO, S::Future, E>>>> {
165        let me = self.project();
166        match ready!(me.make_service.poll_ready_ref(cx)) {
167            Ok(()) => (),
168            Err(e) => {
169                trace!("make_service closed");
170                return Poll::Ready(Some(Err(crate::Error::new_user_make_service(e))));
171            }
172        }
173
174        if let Some(item) = ready!(me.incoming.poll_accept(cx)) {
175            let io = item.map_err(crate::Error::new_accept)?;
176            let new_fut = me.make_service.make_service_ref(&io);
177            Poll::Ready(Some(Ok(Connecting {
178                future: new_fut,
179                io: Some(io),
180                protocol: me.protocol.clone(),
181            })))
182        } else {
183            Poll::Ready(None)
184        }
185    }
186
187    pub(super) fn poll_watch<W>(
188        mut self: Pin<&mut Self>,
189        cx: &mut task::Context<'_>,
190        watcher: &W,
191    ) -> Poll<crate::Result<()>>
192    where
193        E: NewSvcExec<IO, S::Future, S::Service, E, W>,
194        W: Watcher<IO, S::Service, E>,
195    {
196        loop {
197            if let Some(connecting) = ready!(self.as_mut().poll_next_(cx)?) {
198                let fut = NewSvcTask::new(connecting, watcher.clone());
199                self.as_mut().project().protocol.exec.execute_new_svc(fut);
200            } else {
201                return Poll::Ready(Ok(()));
202            }
203        }
204    }
205}
206
207#[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
208impl<I, IO, IE, S, B, E> Future for Server<I, S, E>
209where
210    I: Accept<Conn = IO, Error = IE>,
211    IE: Into<Box<dyn StdError + Send + Sync>>,
212    IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
213    S: MakeServiceRef<IO, Body, ResBody = B>,
214    S::Error: Into<Box<dyn StdError + Send + Sync>>,
215    B: HttpBody + 'static,
216    B::Error: Into<Box<dyn StdError + Send + Sync>>,
217    E: ConnStreamExec<<S::Service as HttpService<Body>>::Future, B>,
218    E: NewSvcExec<IO, S::Future, S::Service, E, NoopWatcher>,
219{
220    type Output = crate::Result<()>;
221
222    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
223        self.poll_watch(cx, &NoopWatcher)
224    }
225}
226
227impl<I: fmt::Debug, S: fmt::Debug> fmt::Debug for Server<I, S> {
228    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
229        let mut st = f.debug_struct("Server");
230        st.field("listener", &self.incoming);
231        st.finish()
232    }
233}
234
235// ===== impl Builder =====
236
237#[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
238impl<I, E> Builder<I, E> {
239    /// Start a new builder, wrapping an incoming stream and low-level options.
240    ///
241    /// For a more convenient constructor, see [`Server::bind`](Server::bind).
242    pub fn new(incoming: I, protocol: Http_<E>) -> Self {
243        Builder { incoming, protocol }
244    }
245
246    /// Sets whether to use keep-alive for HTTP/1 connections.
247    ///
248    /// Default is `true`.
249    #[cfg(feature = "http1")]
250    #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
251    pub fn http1_keepalive(mut self, val: bool) -> Self {
252        self.protocol.http1_keep_alive(val);
253        self
254    }
255
256    /// Set whether HTTP/1 connections should support half-closures.
257    ///
258    /// Clients can chose to shutdown their write-side while waiting
259    /// for the server to respond. Setting this to `true` will
260    /// prevent closing the connection immediately if `read`
261    /// detects an EOF in the middle of a request.
262    ///
263    /// Default is `false`.
264    #[cfg(feature = "http1")]
265    #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
266    pub fn http1_half_close(mut self, val: bool) -> Self {
267        self.protocol.http1_half_close(val);
268        self
269    }
270
271    /// Set the maximum buffer size.
272    ///
273    /// Default is ~ 400kb.
274    #[cfg(feature = "http1")]
275    #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
276    pub fn http1_max_buf_size(mut self, val: usize) -> Self {
277        self.protocol.max_buf_size(val);
278        self
279    }
280
281    // Sets whether to bunch up HTTP/1 writes until the read buffer is empty.
282    //
283    // This isn't really desirable in most cases, only really being useful in
284    // silly pipeline benchmarks.
285    #[doc(hidden)]
286    #[cfg(feature = "http1")]
287    pub fn http1_pipeline_flush(mut self, val: bool) -> Self {
288        self.protocol.pipeline_flush(val);
289        self
290    }
291
292    /// Set whether HTTP/1 connections should try to use vectored writes,
293    /// or always flatten into a single buffer.
294    ///
295    /// Note that setting this to false may mean more copies of body data,
296    /// but may also improve performance when an IO transport doesn't
297    /// support vectored writes well, such as most TLS implementations.
298    ///
299    /// Setting this to true will force hyper to use queued strategy
300    /// which may eliminate unnecessary cloning on some TLS backends
301    ///
302    /// Default is `auto`. In this mode hyper will try to guess which
303    /// mode to use
304    #[cfg(feature = "http1")]
305    pub fn http1_writev(mut self, enabled: bool) -> Self {
306        self.protocol.http1_writev(enabled);
307        self
308    }
309
310    /// Set whether HTTP/1 connections will write header names as title case at
311    /// the socket level.
312    ///
313    /// Note that this setting does not affect HTTP/2.
314    ///
315    /// Default is false.
316    #[cfg(feature = "http1")]
317    #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
318    pub fn http1_title_case_headers(mut self, val: bool) -> Self {
319        self.protocol.http1_title_case_headers(val);
320        self
321    }
322
323    /// Set whether to support preserving original header cases.
324    ///
325    /// Currently, this will record the original cases received, and store them
326    /// in a private extension on the `Request`. It will also look for and use
327    /// such an extension in any provided `Response`.
328    ///
329    /// Since the relevant extension is still private, there is no way to
330    /// interact with the original cases. The only effect this can have now is
331    /// to forward the cases in a proxy-like fashion.
332    ///
333    /// Note that this setting does not affect HTTP/2.
334    ///
335    /// Default is false.
336    #[cfg(feature = "http1")]
337    #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
338    pub fn http1_preserve_header_case(mut self, val: bool) -> Self {
339        self.protocol.http1_preserve_header_case(val);
340        self
341    }
342
343    /// Set a timeout for reading client request headers. If a client does not
344    /// transmit the entire header within this time, the connection is closed.
345    ///
346    /// Default is None.
347    #[cfg(all(feature = "http1", feature = "runtime"))]
348    #[cfg_attr(docsrs, doc(cfg(all(feature = "http1", feature = "runtime"))))]
349    pub fn http1_header_read_timeout(mut self, read_timeout: Duration) -> Self {
350        self.protocol.http1_header_read_timeout(read_timeout);
351        self
352    }
353
354    /// Sets whether HTTP/1 is required.
355    ///
356    /// Default is `false`.
357    #[cfg(feature = "http1")]
358    #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
359    pub fn http1_only(mut self, val: bool) -> Self {
360        self.protocol.http1_only(val);
361        self
362    }
363
364    /// Sets whether HTTP/2 is required.
365    ///
366    /// Default is `false`.
367    #[cfg(feature = "http2")]
368    #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
369    pub fn http2_only(mut self, val: bool) -> Self {
370        self.protocol.http2_only(val);
371        self
372    }
373
374    /// Sets the [`SETTINGS_INITIAL_WINDOW_SIZE`][spec] option for HTTP2
375    /// stream-level flow control.
376    ///
377    /// Passing `None` will do nothing.
378    ///
379    /// If not set, hyper will use a default.
380    ///
381    /// [spec]: https://http2.github.io/http2-spec/#SETTINGS_INITIAL_WINDOW_SIZE
382    #[cfg(feature = "http2")]
383    #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
384    pub fn http2_initial_stream_window_size(mut self, sz: impl Into<Option<u32>>) -> Self {
385        self.protocol.http2_initial_stream_window_size(sz.into());
386        self
387    }
388
389    /// Sets the max connection-level flow control for HTTP2
390    ///
391    /// Passing `None` will do nothing.
392    ///
393    /// If not set, hyper will use a default.
394    #[cfg(feature = "http2")]
395    #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
396    pub fn http2_initial_connection_window_size(mut self, sz: impl Into<Option<u32>>) -> Self {
397        self.protocol
398            .http2_initial_connection_window_size(sz.into());
399        self
400    }
401
402    /// Sets whether to use an adaptive flow control.
403    ///
404    /// Enabling this will override the limits set in
405    /// `http2_initial_stream_window_size` and
406    /// `http2_initial_connection_window_size`.
407    #[cfg(feature = "http2")]
408    #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
409    pub fn http2_adaptive_window(mut self, enabled: bool) -> Self {
410        self.protocol.http2_adaptive_window(enabled);
411        self
412    }
413
414    /// Sets the maximum frame size to use for HTTP2.
415    ///
416    /// Passing `None` will do nothing.
417    ///
418    /// If not set, hyper will use a default.
419    #[cfg(feature = "http2")]
420    #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
421    pub fn http2_max_frame_size(mut self, sz: impl Into<Option<u32>>) -> Self {
422        self.protocol.http2_max_frame_size(sz);
423        self
424    }
425
426    /// Sets the [`SETTINGS_MAX_CONCURRENT_STREAMS`][spec] option for HTTP2
427    /// connections.
428    ///
429    /// Default is no limit (`std::u32::MAX`). Passing `None` will do nothing.
430    ///
431    /// [spec]: https://http2.github.io/http2-spec/#SETTINGS_MAX_CONCURRENT_STREAMS
432    #[cfg(feature = "http2")]
433    #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
434    pub fn http2_max_concurrent_streams(mut self, max: impl Into<Option<u32>>) -> Self {
435        self.protocol.http2_max_concurrent_streams(max.into());
436        self
437    }
438
439    /// Sets an interval for HTTP2 Ping frames should be sent to keep a
440    /// connection alive.
441    ///
442    /// Pass `None` to disable HTTP2 keep-alive.
443    ///
444    /// Default is currently disabled.
445    ///
446    /// # Cargo Feature
447    ///
448    /// Requires the `runtime` cargo feature to be enabled.
449    #[cfg(all(feature = "runtime", feature = "http2"))]
450    #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
451    pub fn http2_keep_alive_interval(mut self, interval: impl Into<Option<Duration>>) -> Self {
452        self.protocol.http2_keep_alive_interval(interval);
453        self
454    }
455
456    /// Sets a timeout for receiving an acknowledgement of the keep-alive ping.
457    ///
458    /// If the ping is not acknowledged within the timeout, the connection will
459    /// be closed. Does nothing if `http2_keep_alive_interval` is disabled.
460    ///
461    /// Default is 20 seconds.
462    ///
463    /// # Cargo Feature
464    ///
465    /// Requires the `runtime` cargo feature to be enabled.
466    #[cfg(all(feature = "runtime", feature = "http2"))]
467    #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
468    pub fn http2_keep_alive_timeout(mut self, timeout: Duration) -> Self {
469        self.protocol.http2_keep_alive_timeout(timeout);
470        self
471    }
472
473    /// Set the maximum write buffer size for each HTTP/2 stream.
474    ///
475    /// Default is currently ~400KB, but may change.
476    ///
477    /// # Panics
478    ///
479    /// The value must be no larger than `u32::MAX`.
480    #[cfg(feature = "http2")]
481    #[cfg_attr(docsrs, doc(cfg(feature = "http2")))]
482    pub fn http2_max_send_buf_size(mut self, max: usize) -> Self {
483        self.protocol.http2_max_send_buf_size(max);
484        self
485    }
486
487    /// Enables the [extended CONNECT protocol].
488    ///
489    /// [extended CONNECT protocol]: https://datatracker.ietf.org/doc/html/rfc8441#section-4
490    #[cfg(feature = "http2")]
491    pub fn http2_enable_connect_protocol(mut self) -> Self {
492        self.protocol.http2_enable_connect_protocol();
493        self
494    }
495
496    /// Sets the `Executor` to deal with connection tasks.
497    ///
498    /// Default is `tokio::spawn`.
499    pub fn executor<E2>(self, executor: E2) -> Builder<I, E2> {
500        Builder {
501            incoming: self.incoming,
502            protocol: self.protocol.with_executor(executor),
503        }
504    }
505
506    /// Consume this `Builder`, creating a [`Server`](Server).
507    ///
508    /// # Example
509    ///
510    /// ```
511    /// # #[cfg(feature = "tcp")]
512    /// # async fn run() {
513    /// use hyper::{Body, Error, Response, Server};
514    /// use hyper::service::{make_service_fn, service_fn};
515    ///
516    /// // Construct our SocketAddr to listen on...
517    /// let addr = ([127, 0, 0, 1], 3000).into();
518    ///
519    /// // And a MakeService to handle each connection...
520    /// let make_svc = make_service_fn(|_| async {
521    ///     Ok::<_, Error>(service_fn(|_req| async {
522    ///         Ok::<_, Error>(Response::new(Body::from("Hello World")))
523    ///     }))
524    /// });
525    ///
526    /// // Then bind and serve...
527    /// let server = Server::bind(&addr)
528    ///     .serve(make_svc);
529    ///
530    /// // Run forever-ish...
531    /// if let Err(err) = server.await {
532    ///     eprintln!("server error: {}", err);
533    /// }
534    /// # }
535    /// ```
536    pub fn serve<S, B>(self, make_service: S) -> Server<I, S, E>
537    where
538        I: Accept,
539        I::Error: Into<Box<dyn StdError + Send + Sync>>,
540        I::Conn: AsyncRead + AsyncWrite + Unpin + Send + 'static,
541        S: MakeServiceRef<I::Conn, Body, ResBody = B>,
542        S::Error: Into<Box<dyn StdError + Send + Sync>>,
543        B: HttpBody + 'static,
544        B::Error: Into<Box<dyn StdError + Send + Sync>>,
545        E: NewSvcExec<I::Conn, S::Future, S::Service, E, NoopWatcher>,
546        E: ConnStreamExec<<S::Service as HttpService<Body>>::Future, B>,
547    {
548        Server {
549            incoming: self.incoming,
550            make_service,
551            protocol: self.protocol.clone(),
552        }
553    }
554}
555
556#[cfg(feature = "tcp")]
557#[cfg_attr(
558    docsrs,
559    doc(cfg(all(feature = "tcp", any(feature = "http1", feature = "http2"))))
560)]
561impl<E> Builder<AddrIncoming, E> {
562    /// Set whether TCP keepalive messages are enabled on accepted connections.
563    ///
564    /// If `None` is specified, keepalive is disabled, otherwise the duration
565    /// specified will be the time to remain idle before sending TCP keepalive
566    /// probes.
567    pub fn tcp_keepalive(mut self, keepalive: Option<Duration>) -> Self {
568        self.incoming.set_keepalive(keepalive);
569        self
570    }
571
572    /// Set the value of `TCP_NODELAY` option for accepted connections.
573    pub fn tcp_nodelay(mut self, enabled: bool) -> Self {
574        self.incoming.set_nodelay(enabled);
575        self
576    }
577
578    /// Set whether to sleep on accept errors.
579    ///
580    /// A possible scenario is that the process has hit the max open files
581    /// allowed, and so trying to accept a new connection will fail with
582    /// EMFILE. In some cases, it's preferable to just wait for some time, if
583    /// the application will likely close some files (or connections), and try
584    /// to accept the connection again. If this option is true, the error will
585    /// be logged at the error level, since it is still a big deal, and then
586    /// the listener will sleep for 1 second.
587    ///
588    /// In other cases, hitting the max open files should be treat similarly
589    /// to being out-of-memory, and simply error (and shutdown). Setting this
590    /// option to false will allow that.
591    ///
592    /// For more details see [`AddrIncoming::set_sleep_on_errors`]
593    pub fn tcp_sleep_on_accept_errors(mut self, val: bool) -> Self {
594        self.incoming.set_sleep_on_errors(val);
595        self
596    }
597}
598
599// Used by `Server` to optionally watch a `Connection` future.
600//
601// The regular `hyper::Server` just uses a `NoopWatcher`, which does
602// not need to watch anything, and so returns the `Connection` untouched.
603//
604// The `Server::with_graceful_shutdown` needs to keep track of all active
605// connections, and signal that they start to shutdown when prompted, so
606// it has a `GracefulWatcher` implementation to do that.
607pub trait Watcher<I, S: HttpService<Body>, E>: Clone {
608    type Future: Future<Output = crate::Result<()>>;
609
610    fn watch(&self, conn: UpgradeableConnection<I, S, E>) -> Self::Future;
611}
612
613#[allow(missing_debug_implementations)]
614#[derive(Copy, Clone)]
615pub struct NoopWatcher;
616
617impl<I, S, E> Watcher<I, S, E> for NoopWatcher
618where
619    I: AsyncRead + AsyncWrite + Unpin + Send + 'static,
620    S: HttpService<Body>,
621    E: ConnStreamExec<S::Future, S::ResBody>,
622    S::ResBody: 'static,
623    <S::ResBody as HttpBody>::Error: Into<Box<dyn StdError + Send + Sync>>,
624{
625    type Future = UpgradeableConnection<I, S, E>;
626
627    fn watch(&self, conn: UpgradeableConnection<I, S, E>) -> Self::Future {
628        conn
629    }
630}
631
632// used by exec.rs
633pub(crate) mod new_svc {
634    use std::error::Error as StdError;
635    use tokio::io::{AsyncRead, AsyncWrite};
636    use tracing::debug;
637
638    use super::{Connecting, Watcher};
639    use crate::body::{Body, HttpBody};
640    use crate::common::exec::ConnStreamExec;
641    use crate::common::{task, Future, Pin, Poll, Unpin};
642    use crate::service::HttpService;
643    use pin_project_lite::pin_project;
644
645    // This is a `Future<Item=(), Error=()>` spawned to an `Executor` inside
646    // the `Server`. By being a nameable type, we can be generic over the
647    // user's `Service::Future`, and thus an `Executor` can execute it.
648    //
649    // Doing this allows for the server to conditionally require `Send` futures,
650    // depending on the `Executor` configured.
651    //
652    // Users cannot import this type, nor the associated `NewSvcExec`. Instead,
653    // a blanket implementation for `Executor<impl Future>` is sufficient.
654
655    pin_project! {
656        #[allow(missing_debug_implementations)]
657        pub struct NewSvcTask<I, N, S: HttpService<Body>, E, W: Watcher<I, S, E>> {
658            #[pin]
659            state: State<I, N, S, E, W>,
660        }
661    }
662
663    pin_project! {
664        #[project = StateProj]
665        pub(super) enum State<I, N, S: HttpService<Body>, E, W: Watcher<I, S, E>> {
666            Connecting {
667                #[pin]
668                connecting: Connecting<I, N, E>,
669                watcher: W,
670            },
671            Connected {
672                #[pin]
673                future: W::Future,
674            },
675        }
676    }
677
678    impl<I, N, S: HttpService<Body>, E, W: Watcher<I, S, E>> NewSvcTask<I, N, S, E, W> {
679        pub(super) fn new(connecting: Connecting<I, N, E>, watcher: W) -> Self {
680            NewSvcTask {
681                state: State::Connecting {
682                    connecting,
683                    watcher,
684                },
685            }
686        }
687    }
688
689    impl<I, N, S, NE, B, E, W> Future for NewSvcTask<I, N, S, E, W>
690    where
691        I: AsyncRead + AsyncWrite + Unpin + Send + 'static,
692        N: Future<Output = Result<S, NE>>,
693        NE: Into<Box<dyn StdError + Send + Sync>>,
694        S: HttpService<Body, ResBody = B>,
695        B: HttpBody + 'static,
696        B::Error: Into<Box<dyn StdError + Send + Sync>>,
697        E: ConnStreamExec<S::Future, B>,
698        W: Watcher<I, S, E>,
699    {
700        type Output = ();
701
702        fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
703            // If it weren't for needing to name this type so the `Send` bounds
704            // could be projected to the `Serve` executor, this could just be
705            // an `async fn`, and much safer. Woe is me.
706
707            let mut me = self.project();
708            loop {
709                let next = {
710                    match me.state.as_mut().project() {
711                        StateProj::Connecting {
712                            connecting,
713                            watcher,
714                        } => {
715                            let res = ready!(connecting.poll(cx));
716                            let conn = match res {
717                                Ok(conn) => conn,
718                                Err(err) => {
719                                    let err = crate::Error::new_user_make_service(err);
720                                    debug!("connecting error: {}", err);
721                                    return Poll::Ready(());
722                                }
723                            };
724                            let future = watcher.watch(conn.with_upgrades());
725                            State::Connected { future }
726                        }
727                        StateProj::Connected { future } => {
728                            return future.poll(cx).map(|res| {
729                                if let Err(err) = res {
730                                    debug!("connection error: {}", err);
731                                }
732                            });
733                        }
734                    }
735                };
736
737                me.state.set(next);
738            }
739        }
740    }
741}
742
743pin_project! {
744    /// A future building a new `Service` to a `Connection`.
745    ///
746    /// Wraps the future returned from `MakeService` into one that returns
747    /// a `Connection`.
748    #[must_use = "futures do nothing unless polled"]
749    #[derive(Debug)]
750    #[cfg_attr(docsrs, doc(cfg(any(feature = "http1", feature = "http2"))))]
751    pub struct Connecting<I, F, E = Exec> {
752        #[pin]
753        future: F,
754        io: Option<I>,
755        protocol: Http_<E>,
756    }
757}
758
759impl<I, F, S, FE, E, B> Future for Connecting<I, F, E>
760where
761    I: AsyncRead + AsyncWrite + Unpin,
762    F: Future<Output = Result<S, FE>>,
763    S: HttpService<Body, ResBody = B>,
764    B: HttpBody + 'static,
765    B::Error: Into<Box<dyn StdError + Send + Sync>>,
766    E: ConnStreamExec<S::Future, B>,
767{
768    type Output = Result<Connection<I, S, E>, FE>;
769
770    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
771        let mut me = self.project();
772        let service = ready!(me.future.poll(cx))?;
773        let io = Option::take(&mut me.io).expect("polled after complete");
774        Poll::Ready(Ok(me.protocol.serve_connection(io, service)))
775    }
776}
hyper/server/server.rs

hyper/server/
server.rs
