saluki_io/net/server/
http.rs

1//! Basic HTTP server.
2
3use std::{
4    future::Future,
5    pin::Pin,
6    sync::Arc,
7    task::{ready, Context, Poll},
8};
9
10use http::{Request, Response};
11use http_body::Body;
12use hyper::{body::Incoming, service::Service};
13use hyper_util::{
14    rt::{TokioExecutor, TokioIo},
15    server::conn::auto::Builder,
16};
17use rustls::ServerConfig;
18use saluki_common::{
19    sync::shutdown::{ShutdownCoordinator, ShutdownHandle},
20    task::{spawn_traced_named, HandleExt as _},
21};
22use saluki_error::GenericError;
23use saluki_tls::ensure_server_config_fips_compliant;
24use tokio::{pin, runtime::Handle, select, sync::oneshot};
25use tokio_rustls::TlsAcceptor;
26use tracing::{debug, error, info};
27
28use crate::net::listener::ConnectionOrientedListener;
29
30/// An HTTP server.
31pub struct HttpServer<S> {
32    executor: Handle,
33    listener: ConnectionOrientedListener,
34    conn_builder: Builder<TokioExecutor>,
35    service: S,
36    tls_config: Option<ServerConfig>,
37}
38
39impl<S> HttpServer<S> {
40    /// Creates a new `HttpServer` from the given listener and service.
41    ///
42    /// # Panics
43    ///
44    /// This will panic if called outside the context of a Tokio runtime.
45    pub fn from_listener(listener: ConnectionOrientedListener, service: S) -> Self {
46        Self {
47            executor: Handle::current(),
48            listener,
49            conn_builder: Builder::new(TokioExecutor::new()),
50            service,
51            tls_config: None,
52        }
53    }
54
55    /// Sets the TLS configuration for the server.
56    ///
57    /// This will enable TLS for the server, and the server will only accept connections that are encrypted with TLS.
58    ///
59    /// Defaults to TLS being disabled.
60    pub fn with_tls_config(mut self, config: ServerConfig) -> Self {
61        self.tls_config = Some(config);
62        self
63    }
64
65    /// Sets the executor for the server.
66    ///
67    /// This executor will be used for spawning tasks to handle incoming connections, but _not_ for the spawn that accepts
68    /// new connections.
69    ///
70    /// Defaults to the current Tokio runtime at the time `HttpServer::new` is called.
71    pub fn with_executor(mut self, executor: Handle) -> Self {
72        self.executor = executor;
73        self
74    }
75}
76
77impl<S, B> HttpServer<S>
78where
79    S: Service<Request<Incoming>, Response = Response<B>> + Send + Clone + 'static,
80    S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
81    S::Future: Send + 'static,
82    B: Body + Send + 'static,
83    B::Data: Send,
84    B::Error: std::error::Error + Send + Sync,
85{
86    /// Starts the server and listens for incoming connections.
87    ///
88    /// Returns two handles: one for shutting down the server, and one for receiving any errors that occur while the
89    /// server is running.
90    pub fn listen(self) -> (ShutdownCoordinator, ErrorHandle) {
91        let (shutdown_coordinator, shutdown) = ShutdownHandle::paired();
92        let (error_tx, error_rx) = oneshot::channel();
93
94        let Self {
95            executor,
96            mut listener,
97            conn_builder,
98            service,
99            tls_config,
100            ..
101        } = self;
102
103        spawn_traced_named("http-server-acceptor", async move {
104            let maybe_tls_acceptor = match tls_config {
105                Some(mut config) => {
106                    // Allow for HTTP/1.1 and HTTP/2.
107                    config.alpn_protocols.push(b"h2".to_vec());
108                    config.alpn_protocols.push(b"http/1.1".to_vec());
109
110                    if let Err(e) = ensure_server_config_fips_compliant(&mut config) {
111                        let _ = error_tx.send(e);
112                        return;
113                    }
114
115                    Some(TlsAcceptor::from(Arc::new(config)))
116                }
117                None => None,
118            };
119            let tls_enabled = maybe_tls_acceptor.is_some();
120
121            info!(listen_addr = %listener.listen_address(), tls_enabled, "HTTP server started.");
122
123            pin!(shutdown);
124
125            loop {
126                select! {
127                    result = listener.accept() => match result {
128                        Ok(stream) => {
129                            let service = service.clone();
130                            let conn_builder = conn_builder.clone();
131                            let listen_addr = listener.listen_address().clone();
132                            match &maybe_tls_acceptor {
133                                Some(acceptor) => {
134                                    let tls_stream = match acceptor.accept(stream).await {
135                                        Ok(stream) => stream,
136                                        Err(e) => {
137                                            error!(%listen_addr, error = %e, "Failed to complete TLS handshake.");
138                                            continue
139                                        },
140                                    };
141
142                                    executor.spawn_traced_named("http-server-tls-conn-handler", async move {
143                                        if let Err(e) = conn_builder.serve_connection(TokioIo::new(tls_stream), service).await {
144                                            error!(%listen_addr, error = %e, "Failed to serve HTTP connection.");
145                                        }
146                                    });
147                                },
148                                None => {
149                                    executor.spawn_traced_named("http-server-conn-handler", async move {
150                                        if let Err(e) = conn_builder.serve_connection(TokioIo::new(stream), service).await {
151                                            error!(%listen_addr, error = %e, "Failed to serve HTTP connection.");
152                                        }
153                                    });
154                                },
155                            }
156                        },
157                        Err(e) => {
158                            let _ = error_tx.send(e.into());
159                            break;
160                        }
161                    },
162
163                    _ = &mut shutdown => {
164                        debug!(listen_addr = %listener.listen_address(), "Received shutdown signal.");
165                        break;
166                    }
167                }
168            }
169
170            info!(listen_addr = %listener.listen_address(), "HTTP server stopped.");
171        });
172
173        (shutdown_coordinator, ErrorHandle(error_rx))
174    }
175}
176
177/// A future that resolves when [`HttpServer`] encounters an unrecoverable error.
178pub struct ErrorHandle(oneshot::Receiver<GenericError>);
179
180impl Future for ErrorHandle {
181    type Output = Option<GenericError>;
182
183    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
184        match ready!(Pin::new(&mut self.0).poll(cx)) {
185            Ok(err) => Poll::Ready(Some(err)),
186            Err(_) => Poll::Ready(None),
187        }
188    }
189}
saluki_io/net/server/http.rs

saluki_io/net/server/
http.rs
