1 //! HTTP Server 2 //! 3 //! A `Server` is created to listen on a port, parse HTTP requests, and hand 4 //! them off to a `Service`. 5 //! 6 //! There are two levels of APIs provide for constructing HTTP servers: 7 //! 8 //! - The higher-level [`Server`](Server) type. 9 //! - The lower-level [`conn`](conn) module. 10 //! 11 //! # Server 12 //! 13 //! The [`Server`](Server) is main way to start listening for HTTP requests. 14 //! It wraps a listener with a [`MakeService`](crate::service), and then should 15 //! be executed to start serving requests. 16 //! 17 //! [`Server`](Server) accepts connections in both HTTP1 and HTTP2 by default. 18 //! 19 //! ## Example 20 //! 21 //! ```no_run 22 //! use std::convert::Infallible; 23 //! use std::net::SocketAddr; 24 //! use hyper::{Body, Request, Response, Server}; 25 //! use hyper::service::{make_service_fn, service_fn}; 26 //! 27 //! async fn handle(_req: Request<Body>) -> Result<Response<Body>, Infallible> { 28 //! Ok(Response::new(Body::from("Hello World"))) 29 //! } 30 //! 31 //! # #[cfg(feature = "runtime")] 32 //! #[tokio::main] 33 //! async fn main() { 34 //! // Construct our SocketAddr to listen on... 35 //! let addr = SocketAddr::from(([127, 0, 0, 1], 3000)); 36 //! 37 //! // And a MakeService to handle each connection... 38 //! let make_service = make_service_fn(|_conn| async { 39 //! Ok::<_, Infallible>(service_fn(handle)) 40 //! }); 41 //! 42 //! // Then bind and serve... 43 //! let server = Server::bind(&addr).serve(make_service); 44 //! 45 //! // And run forever... 46 //! if let Err(e) = server.await { 47 //! eprintln!("server error: {}", e); 48 //! } 49 //! } 50 //! # #[cfg(not(feature = "runtime"))] 51 //! # fn main() {} 52 //! ``` 53 54 pub mod accept; 55 pub mod conn; 56 mod shutdown; 57 #[cfg(feature = "tcp")] 58 mod tcp; 59 60 use std::error::Error as StdError; 61 use std::fmt; 62 #[cfg(feature = "tcp")] 63 use std::net::{SocketAddr, TcpListener as StdTcpListener}; 64 65 #[cfg(feature = "tcp")] 66 use std::time::Duration; 67 68 use pin_project::pin_project; 69 use tokio::io::{AsyncRead, AsyncWrite}; 70 71 use self::accept::Accept; 72 use crate::body::{Body, HttpBody}; 73 use crate::common::exec::{Exec, H2Exec, NewSvcExec}; 74 use crate::common::{task, Future, Pin, Poll, Unpin}; 75 use crate::service::{HttpService, MakeServiceRef}; 76 // Renamed `Http` as `Http_` for now so that people upgrading don't see an 77 // error that `hyper::server::Http` is private... 78 use self::conn::{Http as Http_, NoopWatcher, SpawnAll}; 79 use self::shutdown::{Graceful, GracefulWatcher}; 80 #[cfg(feature = "tcp")] 81 use self::tcp::AddrIncoming; 82 83 /// A listening HTTP server that accepts connections in both HTTP1 and HTTP2 by default. 84 /// 85 /// `Server` is a `Future` mapping a bound listener with a set of service 86 /// handlers. It is built using the [`Builder`](Builder), and the future 87 /// completes when the server has been shutdown. It should be run by an 88 /// `Executor`. 89 #[pin_project] 90 pub struct Server<I, S, E = Exec> { 91 #[pin] 92 spawn_all: SpawnAll<I, S, E>, 93 } 94 95 /// A builder for a [`Server`](Server). 96 #[derive(Debug)] 97 pub struct Builder<I, E = Exec> { 98 incoming: I, 99 protocol: Http_<E>, 100 } 101 102 // ===== impl Server ===== 103 104 impl<I> Server<I, ()> { 105 /// Starts a [`Builder`](Builder) with the provided incoming stream. builder(incoming: I) -> Builder<I>106 pub fn builder(incoming: I) -> Builder<I> { 107 Builder { 108 incoming, 109 protocol: Http_::new(), 110 } 111 } 112 } 113 114 #[cfg(feature = "tcp")] 115 impl Server<AddrIncoming, ()> { 116 /// Binds to the provided address, and returns a [`Builder`](Builder). 117 /// 118 /// # Panics 119 /// 120 /// This method will panic if binding to the address fails. For a method 121 /// to bind to an address and return a `Result`, see `Server::try_bind`. bind(addr: &SocketAddr) -> Builder<AddrIncoming>122 pub fn bind(addr: &SocketAddr) -> Builder<AddrIncoming> { 123 let incoming = AddrIncoming::new(addr).unwrap_or_else(|e| { 124 panic!("error binding to {}: {}", addr, e); 125 }); 126 Server::builder(incoming) 127 } 128 129 /// Tries to bind to the provided address, and returns a [`Builder`](Builder). try_bind(addr: &SocketAddr) -> crate::Result<Builder<AddrIncoming>>130 pub fn try_bind(addr: &SocketAddr) -> crate::Result<Builder<AddrIncoming>> { 131 AddrIncoming::new(addr).map(Server::builder) 132 } 133 134 /// Create a new instance from a `std::net::TcpListener` instance. from_tcp(listener: StdTcpListener) -> Result<Builder<AddrIncoming>, crate::Error>135 pub fn from_tcp(listener: StdTcpListener) -> Result<Builder<AddrIncoming>, crate::Error> { 136 AddrIncoming::from_std(listener).map(Server::builder) 137 } 138 } 139 140 #[cfg(feature = "tcp")] 141 impl<S, E> Server<AddrIncoming, S, E> { 142 /// Returns the local address that this server is bound to. local_addr(&self) -> SocketAddr143 pub fn local_addr(&self) -> SocketAddr { 144 self.spawn_all.local_addr() 145 } 146 } 147 148 impl<I, IO, IE, S, E, B> Server<I, S, E> 149 where 150 I: Accept<Conn = IO, Error = IE>, 151 IE: Into<Box<dyn StdError + Send + Sync>>, 152 IO: AsyncRead + AsyncWrite + Unpin + Send + 'static, 153 S: MakeServiceRef<IO, Body, ResBody = B>, 154 S::Error: Into<Box<dyn StdError + Send + Sync>>, 155 B: HttpBody + Send + Sync + 'static, 156 B::Error: Into<Box<dyn StdError + Send + Sync>>, 157 E: H2Exec<<S::Service as HttpService<Body>>::Future, B>, 158 E: NewSvcExec<IO, S::Future, S::Service, E, GracefulWatcher>, 159 { 160 /// Prepares a server to handle graceful shutdown when the provided future 161 /// completes. 162 /// 163 /// # Example 164 /// 165 /// ``` 166 /// # fn main() {} 167 /// # #[cfg(feature = "tcp")] 168 /// # async fn run() { 169 /// # use hyper::{Body, Response, Server, Error}; 170 /// # use hyper::service::{make_service_fn, service_fn}; 171 /// # let make_service = make_service_fn(|_| async { 172 /// # Ok::<_, Error>(service_fn(|_req| async { 173 /// # Ok::<_, Error>(Response::new(Body::from("Hello World"))) 174 /// # })) 175 /// # }); 176 /// // Make a server from the previous examples... 177 /// let server = Server::bind(&([127, 0, 0, 1], 3000).into()) 178 /// .serve(make_service); 179 /// 180 /// // Prepare some signal for when the server should start shutting down... 181 /// let (tx, rx) = tokio::sync::oneshot::channel::<()>(); 182 /// let graceful = server 183 /// .with_graceful_shutdown(async { 184 /// rx.await.ok(); 185 /// }); 186 /// 187 /// // Await the `server` receiving the signal... 188 /// if let Err(e) = graceful.await { 189 /// eprintln!("server error: {}", e); 190 /// } 191 /// 192 /// // And later, trigger the signal by calling `tx.send(())`. 193 /// let _ = tx.send(()); 194 /// # } 195 /// ``` with_graceful_shutdown<F>(self, signal: F) -> Graceful<I, S, F, E> where F: Future<Output = ()>,196 pub fn with_graceful_shutdown<F>(self, signal: F) -> Graceful<I, S, F, E> 197 where 198 F: Future<Output = ()>, 199 { 200 Graceful::new(self.spawn_all, signal) 201 } 202 } 203 204 impl<I, IO, IE, S, B, E> Future for Server<I, S, E> 205 where 206 I: Accept<Conn = IO, Error = IE>, 207 IE: Into<Box<dyn StdError + Send + Sync>>, 208 IO: AsyncRead + AsyncWrite + Unpin + Send + 'static, 209 S: MakeServiceRef<IO, Body, ResBody = B>, 210 S::Error: Into<Box<dyn StdError + Send + Sync>>, 211 B: HttpBody + 'static, 212 B::Error: Into<Box<dyn StdError + Send + Sync>>, 213 E: H2Exec<<S::Service as HttpService<Body>>::Future, B>, 214 E: NewSvcExec<IO, S::Future, S::Service, E, NoopWatcher>, 215 { 216 type Output = crate::Result<()>; 217 poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output>218 fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> { 219 self.project().spawn_all.poll_watch(cx, &NoopWatcher) 220 } 221 } 222 223 impl<I: fmt::Debug, S: fmt::Debug> fmt::Debug for Server<I, S> { fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result224 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 225 f.debug_struct("Server") 226 .field("listener", &self.spawn_all.incoming_ref()) 227 .finish() 228 } 229 } 230 231 // ===== impl Builder ===== 232 233 impl<I, E> Builder<I, E> { 234 /// Start a new builder, wrapping an incoming stream and low-level options. 235 /// 236 /// For a more convenient constructor, see [`Server::bind`](Server::bind). new(incoming: I, protocol: Http_<E>) -> Self237 pub fn new(incoming: I, protocol: Http_<E>) -> Self { 238 Builder { incoming, protocol } 239 } 240 241 /// Sets whether to use keep-alive for HTTP/1 connections. 242 /// 243 /// Default is `true`. http1_keepalive(mut self, val: bool) -> Self244 pub fn http1_keepalive(mut self, val: bool) -> Self { 245 self.protocol.http1_keep_alive(val); 246 self 247 } 248 249 /// Set whether HTTP/1 connections should support half-closures. 250 /// 251 /// Clients can chose to shutdown their write-side while waiting 252 /// for the server to respond. Setting this to `true` will 253 /// prevent closing the connection immediately if `read` 254 /// detects an EOF in the middle of a request. 255 /// 256 /// Default is `false`. http1_half_close(mut self, val: bool) -> Self257 pub fn http1_half_close(mut self, val: bool) -> Self { 258 self.protocol.http1_half_close(val); 259 self 260 } 261 262 /// Set the maximum buffer size. 263 /// 264 /// Default is ~ 400kb. http1_max_buf_size(mut self, val: usize) -> Self265 pub fn http1_max_buf_size(mut self, val: usize) -> Self { 266 self.protocol.max_buf_size(val); 267 self 268 } 269 270 // Sets whether to bunch up HTTP/1 writes until the read buffer is empty. 271 // 272 // This isn't really desirable in most cases, only really being useful in 273 // silly pipeline benchmarks. 274 #[doc(hidden)] http1_pipeline_flush(mut self, val: bool) -> Self275 pub fn http1_pipeline_flush(mut self, val: bool) -> Self { 276 self.protocol.pipeline_flush(val); 277 self 278 } 279 280 /// Set whether HTTP/1 connections should try to use vectored writes, 281 /// or always flatten into a single buffer. 282 /// 283 /// # Note 284 /// 285 /// Setting this to `false` may mean more copies of body data, 286 /// but may also improve performance when an IO transport doesn't 287 /// support vectored writes well, such as most TLS implementations. 288 /// 289 /// Setting this to true will force hyper to use queued strategy 290 /// which may eliminate unnecessary cloning on some TLS backends 291 /// 292 /// Default is `auto`. In this mode hyper will try to guess which 293 /// mode to use http1_writev(mut self, val: bool) -> Self294 pub fn http1_writev(mut self, val: bool) -> Self { 295 self.protocol.http1_writev(val); 296 self 297 } 298 299 /// Sets whether HTTP/1 is required. 300 /// 301 /// Default is `false`. http1_only(mut self, val: bool) -> Self302 pub fn http1_only(mut self, val: bool) -> Self { 303 self.protocol.http1_only(val); 304 self 305 } 306 307 /// Sets whether HTTP/2 is required. 308 /// 309 /// Default is `false`. http2_only(mut self, val: bool) -> Self310 pub fn http2_only(mut self, val: bool) -> Self { 311 self.protocol.http2_only(val); 312 self 313 } 314 315 /// Sets the [`SETTINGS_INITIAL_WINDOW_SIZE`][spec] option for HTTP2 316 /// stream-level flow control. 317 /// 318 /// Passing `None` will do nothing. 319 /// 320 /// If not set, hyper will use a default. 321 /// 322 /// [spec]: https://http2.github.io/http2-spec/#SETTINGS_INITIAL_WINDOW_SIZE http2_initial_stream_window_size(mut self, sz: impl Into<Option<u32>>) -> Self323 pub fn http2_initial_stream_window_size(mut self, sz: impl Into<Option<u32>>) -> Self { 324 self.protocol.http2_initial_stream_window_size(sz.into()); 325 self 326 } 327 328 /// Sets the max connection-level flow control for HTTP2 329 /// 330 /// Passing `None` will do nothing. 331 /// 332 /// If not set, hyper will use a default. http2_initial_connection_window_size(mut self, sz: impl Into<Option<u32>>) -> Self333 pub fn http2_initial_connection_window_size(mut self, sz: impl Into<Option<u32>>) -> Self { 334 self.protocol 335 .http2_initial_connection_window_size(sz.into()); 336 self 337 } 338 339 /// Sets whether to use an adaptive flow control. 340 /// 341 /// Enabling this will override the limits set in 342 /// `http2_initial_stream_window_size` and 343 /// `http2_initial_connection_window_size`. http2_adaptive_window(mut self, enabled: bool) -> Self344 pub fn http2_adaptive_window(mut self, enabled: bool) -> Self { 345 self.protocol.http2_adaptive_window(enabled); 346 self 347 } 348 349 /// Sets the maximum frame size to use for HTTP2. 350 /// 351 /// Passing `None` will do nothing. 352 /// 353 /// If not set, hyper will use a default. http2_max_frame_size(mut self, sz: impl Into<Option<u32>>) -> Self354 pub fn http2_max_frame_size(mut self, sz: impl Into<Option<u32>>) -> Self { 355 self.protocol.http2_max_frame_size(sz); 356 self 357 } 358 359 /// Sets the [`SETTINGS_MAX_CONCURRENT_STREAMS`][spec] option for HTTP2 360 /// connections. 361 /// 362 /// Default is no limit (`std::u32::MAX`). Passing `None` will do nothing. 363 /// 364 /// [spec]: https://http2.github.io/http2-spec/#SETTINGS_MAX_CONCURRENT_STREAMS http2_max_concurrent_streams(mut self, max: impl Into<Option<u32>>) -> Self365 pub fn http2_max_concurrent_streams(mut self, max: impl Into<Option<u32>>) -> Self { 366 self.protocol.http2_max_concurrent_streams(max.into()); 367 self 368 } 369 370 /// Sets an interval for HTTP2 Ping frames should be sent to keep a 371 /// connection alive. 372 /// 373 /// Pass `None` to disable HTTP2 keep-alive. 374 /// 375 /// Default is currently disabled. 376 /// 377 /// # Cargo Feature 378 /// 379 /// Requires the `runtime` cargo feature to be enabled. 380 #[cfg(feature = "runtime")] http2_keep_alive_interval(mut self, interval: impl Into<Option<Duration>>) -> Self381 pub fn http2_keep_alive_interval(mut self, interval: impl Into<Option<Duration>>) -> Self { 382 self.protocol.http2_keep_alive_interval(interval); 383 self 384 } 385 386 /// Sets a timeout for receiving an acknowledgement of the keep-alive ping. 387 /// 388 /// If the ping is not acknowledged within the timeout, the connection will 389 /// be closed. Does nothing if `http2_keep_alive_interval` is disabled. 390 /// 391 /// Default is 20 seconds. 392 /// 393 /// # Cargo Feature 394 /// 395 /// Requires the `runtime` cargo feature to be enabled. 396 #[cfg(feature = "runtime")] http2_keep_alive_timeout(mut self, timeout: Duration) -> Self397 pub fn http2_keep_alive_timeout(mut self, timeout: Duration) -> Self { 398 self.protocol.http2_keep_alive_timeout(timeout); 399 self 400 } 401 402 /// Sets the `Executor` to deal with connection tasks. 403 /// 404 /// Default is `tokio::spawn`. executor<E2>(self, executor: E2) -> Builder<I, E2>405 pub fn executor<E2>(self, executor: E2) -> Builder<I, E2> { 406 Builder { 407 incoming: self.incoming, 408 protocol: self.protocol.with_executor(executor), 409 } 410 } 411 412 /// Consume this `Builder`, creating a [`Server`](Server). 413 /// 414 /// # Example 415 /// 416 /// ``` 417 /// # #[cfg(feature = "tcp")] 418 /// # async fn run() { 419 /// use hyper::{Body, Error, Response, Server}; 420 /// use hyper::service::{make_service_fn, service_fn}; 421 /// 422 /// // Construct our SocketAddr to listen on... 423 /// let addr = ([127, 0, 0, 1], 3000).into(); 424 /// 425 /// // And a MakeService to handle each connection... 426 /// let make_svc = make_service_fn(|_| async { 427 /// Ok::<_, Error>(service_fn(|_req| async { 428 /// Ok::<_, Error>(Response::new(Body::from("Hello World"))) 429 /// })) 430 /// }); 431 /// 432 /// // Then bind and serve... 433 /// let server = Server::bind(&addr) 434 /// .serve(make_svc); 435 /// 436 /// // Run forever-ish... 437 /// if let Err(err) = server.await { 438 /// eprintln!("server error: {}", err); 439 /// } 440 /// # } 441 /// ``` serve<S, B>(self, new_service: S) -> Server<I, S, E> where I: Accept, I::Error: Into<Box<dyn StdError + Send + Sync>>, I::Conn: AsyncRead + AsyncWrite + Unpin + Send + 'static, S: MakeServiceRef<I::Conn, Body, ResBody = B>, S::Error: Into<Box<dyn StdError + Send + Sync>>, B: HttpBody + 'static, B::Error: Into<Box<dyn StdError + Send + Sync>>, E: NewSvcExec<I::Conn, S::Future, S::Service, E, NoopWatcher>, E: H2Exec<<S::Service as HttpService<Body>>::Future, B>,442 pub fn serve<S, B>(self, new_service: S) -> Server<I, S, E> 443 where 444 I: Accept, 445 I::Error: Into<Box<dyn StdError + Send + Sync>>, 446 I::Conn: AsyncRead + AsyncWrite + Unpin + Send + 'static, 447 S: MakeServiceRef<I::Conn, Body, ResBody = B>, 448 S::Error: Into<Box<dyn StdError + Send + Sync>>, 449 B: HttpBody + 'static, 450 B::Error: Into<Box<dyn StdError + Send + Sync>>, 451 E: NewSvcExec<I::Conn, S::Future, S::Service, E, NoopWatcher>, 452 E: H2Exec<<S::Service as HttpService<Body>>::Future, B>, 453 { 454 let serve = self.protocol.serve(self.incoming, new_service); 455 let spawn_all = serve.spawn_all(); 456 Server { spawn_all } 457 } 458 } 459 460 #[cfg(feature = "tcp")] 461 impl<E> Builder<AddrIncoming, E> { 462 /// Set whether TCP keepalive messages are enabled on accepted connections. 463 /// 464 /// If `None` is specified, keepalive is disabled, otherwise the duration 465 /// specified will be the time to remain idle before sending TCP keepalive 466 /// probes. tcp_keepalive(mut self, keepalive: Option<Duration>) -> Self467 pub fn tcp_keepalive(mut self, keepalive: Option<Duration>) -> Self { 468 self.incoming.set_keepalive(keepalive); 469 self 470 } 471 472 /// Set the value of `TCP_NODELAY` option for accepted connections. tcp_nodelay(mut self, enabled: bool) -> Self473 pub fn tcp_nodelay(mut self, enabled: bool) -> Self { 474 self.incoming.set_nodelay(enabled); 475 self 476 } 477 478 /// Set whether to sleep on accept errors. 479 /// 480 /// A possible scenario is that the process has hit the max open files 481 /// allowed, and so trying to accept a new connection will fail with 482 /// EMFILE. In some cases, it's preferable to just wait for some time, if 483 /// the application will likely close some files (or connections), and try 484 /// to accept the connection again. If this option is true, the error will 485 /// be logged at the error level, since it is still a big deal, and then 486 /// the listener will sleep for 1 second. 487 /// 488 /// In other cases, hitting the max open files should be treat similarly 489 /// to being out-of-memory, and simply error (and shutdown). Setting this 490 /// option to false will allow that. 491 /// 492 /// For more details see [`AddrIncoming::set_sleep_on_errors`] tcp_sleep_on_accept_errors(mut self, val: bool) -> Self493 pub fn tcp_sleep_on_accept_errors(mut self, val: bool) -> Self { 494 self.incoming.set_sleep_on_errors(val); 495 self 496 } 497 } 498