1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270
use std::future::Future;
use std::net::SocketAddr;
use crate::connection::ConnectionInfo;
use crate::server::configuration::ServerConfiguration;
use crate::server::server_handle::ServerHandle;
use super::IncomingStream;
/// An HTTP server to handle incoming connections for Pavex applications.
/// It handles both HTTP1 and HTTP2 connections.
///
/// # Example
///
/// ```rust
/// use std::net::SocketAddr;
/// use pavex::server::Server;
///
/// # #[derive(Clone)] struct ApplicationState;
/// # async fn router(_req: hyper::Request<hyper::body::Incoming>, _conn_info: Option<pavex::connection::ConnectionInfo>, _state: ApplicationState) -> pavex::response::Response { todo!() }
/// # async fn t() -> std::io::Result<()> {
/// # let application_state = ApplicationState;
/// let addr = SocketAddr::from(([127, 0, 0, 1], 8080));
///
/// Server::new()
/// .bind(addr)
/// .await?
/// // Both the routing function and the application state will usually
/// // be code-generated by Pavex, starting from your `Blueprint`.
/// // You don't have to define them manually!
/// .serve(router, application_state)
/// // The `serve` method returns a `ServerHandle` that you can use to
/// // interact with the server.
/// // Calling `.await` on the handle lets you wait until the server
/// // shuts down.
/// .await;
/// # Ok(())
/// # }
/// ```
///
/// # Configuration
///
/// [`Server::new`] returns a new [`Server`] with default configuration.
/// You can customize the server default settings by creating your own [`ServerConfiguration`]
/// and invoking [`Server::set_config`].
///
/// # Architecture
///
/// By default, [`Server::serve`] creates a worker per CPU core and distributes connection from an
/// acceptor thread using a round-robin strategy.
///
/// Each worker has its own single-threaded [`tokio`] runtime—there is no work stealing across
/// workers.
/// Each worker takes care to invoke your routing and request handling logic, with the help
/// of [`hyper`].
#[must_use = "You must call `serve` on a `Server` to start listening for incoming connections"]
pub struct Server {
config: ServerConfiguration,
incoming: Vec<IncomingStream>,
}
impl Default for Server {
fn default() -> Self {
Self::new()
}
}
impl Server {
/// Create a new [`Server`] with default configuration.
pub fn new() -> Self {
Self {
config: ServerConfiguration::default(),
incoming: Vec::new(),
}
}
/// Configure this [`Server`] according to the values set in the [`ServerConfiguration`]
/// passed as input parameter.
/// It will overwrite any previous configuration set on this [`Server`].
///
/// If you want to retrieve the current configuration, use [`Server::get_config`].
pub fn set_config(mut self, config: ServerConfiguration) -> Self {
self.config = config;
self
}
/// Get a reference to the [`ServerConfiguration`] for this [`Server`].
///
/// If you want to overwrite the existing configuration, use [`Server::set_config`].
pub fn get_config(&self) -> &ServerConfiguration {
&self.config
}
/// Bind the server to the given address: the server will accept incoming connections from this
/// address when started.
/// Binding an address may fail (e.g. if the address is already in use), therefore this method
/// may return an error.
///
/// # Related
///
/// Check out [`Server::listen`] for an alternative binding mechanism as well as a
/// discussion of the pros and cons of [`Server::bind`] vs [`Server::listen`].
///
/// # Note
///
/// A [`Server`] can be bound to multiple addresses: just call this method multiple times with
/// all the addresses you want to bind to.
///
/// # Example: bind one address
///
/// ```rust
/// use std::net::SocketAddr;
/// use pavex::server::Server;
///
/// # async fn t() -> std::io::Result<()> {
/// let addr = SocketAddr::from(([127, 0, 0, 1], 8080));
///
/// Server::new()
/// .bind(addr)
/// .await?
/// # ;
/// // [...]
/// # Ok(())
/// # }
/// ```
///
/// # Example: bind multiple addresses
///
/// ```rust
/// use std::net::SocketAddr;
/// use pavex::server::Server;
///
/// # async fn t() -> std::io::Result<()> {
/// let addr1 = SocketAddr::from(([127, 0, 0, 1], 8080));
/// let addr2 = SocketAddr::from(([127, 0, 0, 1], 4000));
///
/// Server::new()
/// .bind(addr1)
/// .await?
/// .bind(addr2)
/// .await?
/// # ;
/// // [...]
/// # Ok(())
/// # }
/// ````
pub async fn bind(mut self, addr: SocketAddr) -> std::io::Result<Self> {
let incoming = IncomingStream::bind(addr).await?;
self.incoming.push(incoming);
Ok(self)
}
/// Ask the server to process incoming connections from the provided [`IncomingStream`].
///
/// # [`Server::listen`] vs [`Server::bind`]
///
/// [`Server::bind`] only requires you to specify the address you want to listen at. The
/// socket configuration is handled by the [`Server`], with a set of reasonable default
/// parameters. You have no access to the [`IncomingStream`] that gets bound to the address
/// you specified.
///
/// [`Server::listen`], instead, expects an [`IncomingStream`].
/// You are free to configure the socket as you see please and the [`Server`] will just
/// poll it for incoming connections.
/// It also allows you to interact with the bound [`IncomingStream`] directly
///
/// # Example: bind to a random port
///
/// ```rust
/// use std::net::SocketAddr;
/// use pavex::server::{IncomingStream, Server};
///
/// # async fn t() -> std::io::Result<()> {
/// // `0` is a special port: it tells the OS to assign us
/// // a random **unused** port
/// let addr = SocketAddr::from(([127, 0, 0, 1], 0));
/// let incoming = IncomingStream::bind(addr).await?;
/// // We can then retrieve the actual port we were assigned
/// // by the OS.
/// let addr = incoming.local_addr()?.to_owned();
///
/// Server::new()
/// .listen(incoming);
/// # ;
/// // [...]
/// # Ok(())
/// # }
/// ````
///
/// # Example: set a custom socket backlog
///
/// ```rust
/// use std::net::SocketAddr;
/// use socket2::Domain;
/// use pavex::server::{IncomingStream, Server};
///
/// # async fn t() -> std::io::Result<()> {
/// // `0` is a special port: it tells the OS to assign us
/// // a random **unused** port
/// let addr = SocketAddr::from(([127, 0, 0, 1], 0));
///
/// let socket = socket2::Socket::new(
/// Domain::for_address(addr),
/// socket2::Type::STREAM,
/// Some(socket2::Protocol::TCP),
/// )
/// .expect("Failed to create a socket");
/// socket.set_reuse_address(true)?;
/// socket.set_nonblocking(true)?;
/// socket.bind(&addr.into())?;
/// // The custom backlog!
/// socket.listen(2048_i32)?;
///
/// let listener = std::net::TcpListener::from(socket);
/// Server::new()
/// .listen(listener.try_into()?)
/// # ;
/// // [...]
/// # Ok(())
/// # }
/// ````
///
/// # Note
///
/// A [`Server`] can listen to multiple streams of incoming connections: just call this method
/// multiple times!
pub fn listen(mut self, incoming: IncomingStream) -> Self {
self.incoming.push(incoming);
self
}
/// Start listening for incoming connections.
///
/// You must specify:
///
/// - a handler function, which will be called for each incoming request;
/// - the application state, the set of singleton components that will be available to
/// your handler function.
///
/// Both the handler function and the application state are usually code-generated by Pavex
/// starting from your [`Blueprint`](crate::blueprint::Blueprint).
///
/// # Wait for the server to shut down
///
/// `serve` returns a [`ServerHandle`].
/// Calling `.await` on the handle lets you wait until the server shuts down.
///
/// # Panics
///
/// This method will panic if the [`Server`] has no registered source of incoming connections,
/// i.e. if you did not call [`Server::bind`] or [`Server::listen`] before calling `serve`.
pub fn serve<HandlerFuture, ApplicationState>(
self,
handler: fn(
http::Request<hyper::body::Incoming>,
Option<ConnectionInfo>,
ApplicationState,
) -> HandlerFuture,
application_state: ApplicationState,
) -> ServerHandle
where
HandlerFuture: Future<Output = crate::response::Response> + 'static,
ApplicationState: Clone + Send + Sync + 'static,
{
if self.incoming.is_empty() {
panic!("Cannot serve: there is no source of incoming connections. Please call `bind` or `listen` on the server before calling `serve`.");
}
ServerHandle::new(self.config, self.incoming, handler, application_state)
}
}