ntex_server/net/
builder.rs

1use std::{fmt, io, net, sync::Arc};
2
3use ntex_io::Io;
4use ntex_rt::System;
5use ntex_service::{IntoService, Service, cfg::SharedCfg};
6use ntex_util::time::Millis;
7use socket2::{Domain, SockAddr, Socket, Type};
8
9use crate::{NoConfig, Server, ServerAppConfig, WorkerPool};
10
11use super::accept::AcceptLoop;
12use super::config::ServiceConfig;
13use super::factory::{self, FactoryServiceType};
14use super::{Connection, ServerStatus, StreamServer, Token, socket::Listener};
15
16/// Builder for a network server.
17///
18/// Register listeners and their service factories, configure the worker pool,
19/// and call [`run`](Self::run) to start the server.
20pub struct ServerBuilder<Cfg = NoConfig> {
21    name: String,
22    token: Token,
23    backlog: i32,
24    state: Arc<Cfg>,
25    services: Vec<FactoryServiceType<Cfg>>,
26    sockets: Vec<(Token, String, Listener)>,
27    accept: AcceptLoop,
28    pool: WorkerPool,
29}
30
31impl Default for ServerBuilder {
32    fn default() -> Self {
33        Self::new(NoConfig)
34    }
35}
36
37impl<Cfg> ServerBuilder<Cfg>
38where
39    Cfg: ServerAppConfig,
40{
41    #[must_use]
42    /// Creates a server builder with the specified application configuration.
43    pub fn new(cfg: Cfg) -> ServerBuilder<Cfg> {
44        let sys = System::current();
45        let mut accept = AcceptLoop::default();
46        accept.name(sys.name());
47        if sys.testing() {
48            accept.testing();
49        }
50
51        ServerBuilder {
52            accept,
53            name: sys.name().to_string(),
54            token: Token(0),
55            state: Arc::new(cfg),
56            services: Vec::new(),
57            sockets: Vec::new(),
58            backlog: 2048,
59            pool: WorkerPool::default().name(sys.name()),
60        }
61    }
62
63    #[must_use]
64    /// Sets the server name.
65    ///
66    /// The name is also used for the accept and worker thread names. It
67    /// defaults to the current system name.
68    pub fn name<T: AsRef<str>>(mut self, name: T) -> Self {
69        self.name = name.as_ref().to_string();
70        self.accept.name(self.name.as_str());
71        self.pool = self.pool.name(self.name.as_str());
72        self
73    }
74
75    #[must_use]
76    /// Sets the number of worker threads to start.
77    ///
78    /// By default, the server uses the number of available logical CPUs.
79    pub fn workers(mut self, num: usize) -> Self {
80        self.pool = self.pool.workers(num);
81        self
82    }
83
84    #[must_use]
85    /// Sets the maximum number of pending connections.
86    ///
87    /// This refers to the number of clients that can be waiting to be served.
88    /// Exceeding this number results in the client getting an error when
89    /// attempting to connect. It should only affect servers under significant
90    /// load.
91    ///
92    /// Generally set in the 64-2048 range. Default value is 2048.
93    ///
94    /// It applies to listeners created by later [`bind`](Self::bind) and
95    /// [`configure`](Self::configure) calls. It does not affect listeners
96    /// passed to [`listen`](Self::listen) or `listen_uds`.
97    pub fn backlog(mut self, num: i32) -> Self {
98        self.backlog = num;
99        self
100    }
101
102    #[must_use]
103    /// Sets the maximum per-worker number of concurrent connections.
104    ///
105    /// A worker stops taking new connections while it is at this limit. When
106    /// no worker can take a connection, the listeners stop accepting.
107    ///
108    /// The limit is a process-wide setting shared by every server in the
109    /// process. Set it before the server starts, because each worker reads
110    /// it when its first service is created.
111    ///
112    /// The default is 25,600 connections per worker.
113    pub fn max_connections(self, num: usize) -> Self {
114        super::max_concurrent_connections(num);
115        self
116    }
117
118    #[must_use]
119    /// Stops the current ntex runtime after the server has stopped.
120    ///
121    /// By default, "stop runtime" is disabled.
122    pub fn stop_runtime(mut self) -> Self {
123        self.pool = self.pool.stop_runtime();
124        self
125    }
126
127    #[must_use]
128    /// Stops the server when one of the workers fails.
129    ///
130    /// A worker fails when it panics or its service cannot be created. The
131    /// stop is graceful only if [`graceful_shutdown`](Self::graceful_shutdown)
132    /// is enabled. Without this option, a failed worker is restarted.
133    ///
134    /// By default, "stop on panic" is disabled.
135    pub fn stop_on_panic(mut self) -> Self {
136        self.pool = self.pool.stop_on_panic();
137        self
138    }
139
140    #[must_use]
141    /// Disables signal handling.
142    ///
143    /// By default, the server stops on SIGINT, SIGTERM, and SIGQUIT.
144    pub fn disable_signals(mut self) -> Self {
145        self.pool = self.pool.disable_signals();
146        self
147    }
148
149    #[must_use]
150    /// Enables CPU affinity for worker threads.
151    ///
152    /// By default, affinity is disabled.
153    pub fn enable_affinity(mut self) -> Self {
154        self.pool = self.pool.enable_affinity();
155        self
156    }
157
158    #[must_use]
159    /// Enables graceful shutdown on SIGQUIT, fatal signals, and panics.
160    ///
161    /// When enabled, SIGQUIT, SIGSEGV, SIGABRT, application panics, and
162    /// worker failures with "stop on panic" stop the server gracefully.
163    /// SIGTERM always stops gracefully and SIGINT always stops immediately.
164    ///
165    /// By default, these events stop the server immediately.
166    pub fn graceful_shutdown(mut self) -> Self {
167        self.pool = self.pool.graceful_shutdown();
168        self
169    }
170
171    #[must_use]
172    /// Timeout for graceful worker shutdown.
173    ///
174    /// After receiving a stop signal, workers have this much time to finish
175    /// serving requests. Workers that are still alive after the timeout are
176    /// forcefully dropped.
177    ///
178    /// This bounds the worker as a whole, not an individual connection. Each
179    /// connection is bound separately by `IoConfig::set_shutdown_timeout`, so
180    /// this value should leave room for the connections a worker is still
181    /// draining to shut down themselves.
182    ///
183    /// By default, the timeout is set to 30 seconds.
184    pub fn graceful_shutdown_timeout<T: Into<Millis>>(mut self, timeout: T) -> Self {
185        self.pool = self.pool.graceful_shutdown_timeout(timeout);
186        self
187    }
188
189    #[must_use]
190    /// Sets the server status handler.
191    ///
192    /// The handler runs on the accept thread. It receives
193    /// [`ServerStatus::Ready`] when the listeners resume accepting and
194    /// [`ServerStatus::NotReady`] when they pause. The same status may be
195    /// reported more than once.
196    pub fn status_handler<F>(mut self, handler: F) -> Self
197    where
198        F: FnMut(ServerStatus) + Send + 'static,
199    {
200        self.accept.set_status_handler(handler);
201        self
202    }
203
204    /// Runs asynchronous configuration as part of the server building
205    /// process.
206    ///
207    /// Listeners registered on the [`ServiceConfig`] are added to the server.
208    /// Services for them are attached per worker in
209    /// [`ServiceConfig::on_worker_start`]. This is useful for moving parts of
210    /// the configuration to a different module or library.
211    pub async fn configure<F>(mut self, f: F) -> io::Result<Self>
212    where
213        F: AsyncFnOnce(ServiceConfig<Cfg>) -> io::Result<()>,
214    {
215        let cfg = ServiceConfig::new(self.token, self.backlog);
216
217        f(cfg.clone()).await?;
218
219        let (token, sockets, factory) = cfg.into_factory();
220        self.token = token;
221        self.sockets.extend(sockets);
222        self.services.push(factory);
223
224        Ok(self)
225    }
226
227    #[allow(clippy::needless_pass_by_value)]
228    /// Binds TCP listeners and registers a service factory.
229    ///
230    /// A listener is created for every address resolved from `addr`. Binding
231    /// succeeds if at least one of them binds; addresses that fail to bind
232    /// are skipped.
233    ///
234    /// `cfg` is the I/O configuration for accepted connections. `factory` is
235    /// called once per worker with that worker's application state and
236    /// returns the connection service.
237    pub fn bind<F, S, I>(
238        mut self,
239        name: impl AsRef<str>,
240        addr: impl net::ToSocketAddrs,
241        cfg: impl Into<SharedCfg>,
242        factory: F,
243    ) -> io::Result<Self>
244    where
245        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
246        S: Service<Cfg::State, Io> + 'static,
247        I: IntoService<S, Cfg::State, Io> + 'static,
248    {
249        let cfg = cfg.into();
250        let sockets = bind_addr(addr, self.backlog)?;
251
252        let mut tokens = Vec::new();
253        for lst in sockets {
254            let token = self.token.next();
255            self.sockets
256                .push((token, name.as_ref().to_string(), Listener::from_tcp(lst)));
257            tokens.push((token, cfg.clone()));
258        }
259
260        self.services.push(factory::create_factory_service(
261            name.as_ref().to_string(),
262            tokens,
263            factory,
264        ));
265
266        Ok(self)
267    }
268
269    #[cfg(unix)]
270    /// Binds a Unix domain socket and registers a service factory.
271    ///
272    /// Any existing file at `addr` is removed before binding. The socket file
273    /// is removed again when the server stops. See [`bind`](Self::bind) for
274    /// `cfg` and `factory`.
275    pub fn bind_uds<F, I, S>(
276        self,
277        name: impl AsRef<str>,
278        addr: impl AsRef<std::path::Path>,
279        cfg: impl Into<SharedCfg>,
280        factory: F,
281    ) -> io::Result<Self>
282    where
283        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
284        I: IntoService<S, Cfg::State, Io> + 'static,
285        S: Service<Cfg::State, Io> + 'static,
286    {
287        use std::os::unix::net::UnixListener;
288
289        // The path must not exist when we try to bind.
290        // Try to remove it to avoid bind error.
291        if let Err(e) = std::fs::remove_file(addr.as_ref()) {
292            // NotFound is expected and not an issue. Anything else is.
293            if e.kind() != std::io::ErrorKind::NotFound {
294                return Err(e);
295            }
296        }
297
298        let lst = UnixListener::bind(addr)?;
299        self.listen_uds(name, lst, cfg.into(), factory)
300    }
301
302    #[cfg(unix)]
303    /// Registers a service factory for an existing Unix domain listener.
304    ///
305    /// This is useful for socket activation, including listeners acquired
306    /// through systemd. The listener is switched to non-blocking mode. See
307    /// [`bind`](Self::bind) for `cfg` and `factory`.
308    pub fn listen_uds<F, I, S>(
309        mut self,
310        name: impl AsRef<str>,
311        lst: std::os::unix::net::UnixListener,
312        cfg: impl Into<SharedCfg>,
313        factory: F,
314    ) -> io::Result<Self>
315    where
316        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
317        I: IntoService<S, Cfg::State, Io> + 'static,
318        S: Service<Cfg::State, Io> + 'static,
319    {
320        let token = self.token.next();
321        self.services.push(factory::create_factory_service(
322            name.as_ref().to_string(),
323            vec![(token, cfg.into())],
324            factory,
325        ));
326        self.sockets
327            .push((token, name.as_ref().to_string(), Listener::from_uds(lst)));
328        Ok(self)
329    }
330
331    /// Registers a service factory for an existing TCP listener.
332    ///
333    /// The listener is switched to non-blocking mode. See
334    /// [`bind`](Self::bind) for `cfg` and `factory`.
335    pub fn listen<F, S, I>(
336        mut self,
337        name: impl AsRef<str>,
338        lst: net::TcpListener,
339        cfg: impl Into<SharedCfg>,
340        factory: F,
341    ) -> io::Result<Self>
342    where
343        F: AsyncFn(&Cfg::State) -> I + Send + Clone + 'static,
344        S: Service<Cfg::State, Io> + 'static,
345        I: IntoService<S, Cfg::State, Io> + 'static,
346    {
347        let token = self.token.next();
348        self.services.push(factory::create_factory_service(
349            name.as_ref().to_string(),
350            vec![(token, cfg.into())],
351            factory,
352        ));
353        self.sockets
354            .push((token, name.as_ref().to_string(), Listener::from_tcp(lst)));
355        Ok(self)
356    }
357
358    /// Starts processing incoming connections and returns a server controller.
359    ///
360    /// # Panics
361    ///
362    /// Panics if no listener has been registered.
363    pub fn run(self) -> Server<Connection> {
364        assert!(
365            !self.sockets.is_empty(),
366            "Server should have at least one bound socket"
367        );
368        let srv = StreamServer::new(self.accept.notify(), self.state, self.services);
369        let svc = self.pool.run(srv);
370
371        let sockets = self
372            .sockets
373            .into_iter()
374            .map(|sock| {
375                log::info!("Starting \"{}\" service on {}", sock.1, sock.2);
376                (sock.0, sock.2)
377            })
378            .collect();
379        self.accept.start(sockets, svc.clone());
380
381        svc
382    }
383}
384
385impl<Cfg> fmt::Debug for ServerBuilder<Cfg> {
386    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
387        f.debug_struct("ServerBuilder")
388            .field("name", &self.name)
389            .field("token", &self.token)
390            .field("backlog", &self.backlog)
391            .field("sockets", &self.sockets)
392            .field("accept", &self.accept)
393            .field("worker-pool", &self.pool)
394            .finish()
395    }
396}
397
398/// Binds TCP listeners for every address resolved from `addr`.
399///
400/// Succeeds if at least one address binds and returns only the listeners that
401/// bound. Otherwise, returns the last bind error.
402pub fn bind_addr<S: net::ToSocketAddrs>(
403    addr: S,
404    backlog: i32,
405) -> io::Result<Vec<net::TcpListener>> {
406    let mut err = None;
407    let mut succ = false;
408    let mut sockets = Vec::new();
409    for addr in addr.to_socket_addrs()? {
410        match create_tcp_listener(addr, backlog) {
411            Ok(lst) => {
412                succ = true;
413                sockets.push(lst);
414            }
415            Err(e) => err = Some(e),
416        }
417    }
418
419    if succ {
420        Ok(sockets)
421    } else if let Some(e) = err.take() {
422        Err(e)
423    } else {
424        Err(io::Error::new(
425            io::ErrorKind::InvalidInput,
426            "Cannot bind to address.",
427        ))
428    }
429}
430
431/// Creates and binds a TCP listener with the specified listen backlog.
432pub fn create_tcp_listener(addr: net::SocketAddr, backlog: i32) -> io::Result<net::TcpListener> {
433    let builder = match addr {
434        net::SocketAddr::V4(_) => Socket::new(Domain::IPV4, Type::STREAM, None)?,
435        net::SocketAddr::V6(_) => Socket::new(Domain::IPV6, Type::STREAM, None)?,
436    };
437
438    // On Windows, this allows rebinding sockets which are actively in use,
439    // which allows “socket hijacking”, so we explicitly don't set it here.
440    // https://docs.microsoft.com/en-us/windows/win32/winsock/using-so-reuseaddr-and-so-exclusiveaddruse
441    #[cfg(not(windows))]
442    builder.set_reuse_address(true)?;
443
444    builder.bind(&SockAddr::from(addr))?;
445    builder.listen(backlog)?;
446    Ok(net::TcpListener::from(builder))
447}
448
449#[cfg(test)]
450mod tests {
451    use super::*;
452
453    #[test]
454    fn test_bind_addr() {
455        let addrs: Vec<net::SocketAddr> = Vec::new();
456        assert!(bind_addr(&addrs[..], 10).is_err());
457    }
458
459    #[ntex::test]
460    async fn test_debug() {
461        let builder = ServerBuilder::default();
462        assert!(format!("{builder:?}").contains("ServerBuilder"));
463    }
464}
ntex_server/net/builder.rs

ntex_server/net/
builder.rs
