ntex_server/net/
mod.rs

1//! General-purpose network server.
2//!
3//! Use [`build()`] or [`ServerBuilder`] to register TCP or Unix domain socket
4//! services. Each worker receives its own service instance and processes
5//! connections on a single-threaded runtime.
6use std::sync::atomic::{AtomicUsize, Ordering};
7
8use ntex_util::services::Counter;
9
10mod accept;
11mod builder;
12mod config;
13mod factory;
14mod service;
15mod socket;
16mod test;
17
18pub use crate::{
19    Affinity, ListenerMetrics, LoadBalancing, NoConfig, RuntimeMetrics, ServerAppConfig,
20    ServerMetrics, ServerStatus, WorkerMetrics,
21};
22
23pub use self::accept::{AcceptLoop, AcceptNotify, AcceptorCommand};
24pub use self::builder::{ServerBuilder, bind_addr, create_tcp_listener};
25pub use self::config::{ServiceConfig, ServiceRuntime};
26pub use self::service::StreamServer;
27pub use self::socket::{Connection, Stream};
28pub use self::test::{TestServer, TestServerBuilder, build_test_server, test_server};
29
30/// Controller for a running network server.
31pub type Server = crate::Server<Connection>;
32
33/// Identifier assigned to a registered listener.
34#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
35pub struct Token(usize);
36
37impl Token {
38    #[must_use]
39    #[allow(clippy::should_implement_trait)]
40    /// Returns the current token and advances this value to the next token.
41    pub fn next(&mut self) -> Token {
42        let token = Token(self.0);
43        self.0 += 1;
44        token
45    }
46}
47
48/// Creates a server builder with no application configuration.
49pub fn build() -> ServerBuilder {
50    ServerBuilder::default()
51}
52
53/// Creates a server builder with application configuration.
54pub fn build_with_config<Cfg>(state: Cfg) -> ServerBuilder<Cfg>
55where
56    Cfg: ServerAppConfig,
57{
58    ServerBuilder::new(state)
59}
60
61static MAX_CONNS: AtomicUsize = AtomicUsize::new(25600);
62
63thread_local! {
64    static MAX_CONNS_COUNTER: Counter = Counter::new(MAX_CONNS.load(Ordering::Relaxed));
65}
66
67/// Sets the maximum per-worker number of concurrent connections.
68///
69/// By default, the limit is 25,600 connections per worker.
70pub(super) fn max_concurrent_connections(num: usize) {
71    MAX_CONNS.store(num, Ordering::Relaxed);
72    MAX_CONNS_COUNTER.with(|conns| conns.set_capacity(num));
73}
74
75pub(super) fn num_connections() -> usize {
76    MAX_CONNS_COUNTER.with(Counter::total)
77}
78
79#[cfg(test)]
80mod tests {
81    use super::*;
82
83    /// STATEMENT: Network server diagnostics describe listener configuration and worker services.
84    ///
85    /// HOW:
86    /// - Configure one named listener and check its token, backlog and socket name in diagnostics.
87    /// - Build the stream server and worker service, checking the factory count and token routing.
88    #[ntex::test]
89    async fn debug_network_services() {
90        use crate::ServerConfiguration;
91        use ntex_service::cfg::SharedCfg;
92
93        let config = ServiceConfig::<NoConfig>::new(Token(7), 64);
94        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
95        config.listen("debug-listener", listener);
96        config.on_worker_start(async |runtime| {
97            runtime.service(
98                "debug-listener",
99                SharedCfg::default(),
100                ntex_service::fn_service(async |_: ntex_io::Io| Ok::<_, ()>(())),
101            );
102            Ok(())
103        });
104        let debug = format!("{config:?}");
105        assert!(debug.contains("ServiceConfigInner { token: Token(8), backlog: 64, sockets: ["));
106        assert!(debug.contains("name: \"debug-listener\""));
107        assert!(debug.contains("Token(7)"));
108
109        let (_, _sockets, factory) = config.into_factory();
110        let accept = AcceptLoop::new();
111        let server = StreamServer::new(
112            accept.notify(),
113            std::sync::Arc::new(NoConfig),
114            vec![factory],
115        );
116        assert_eq!(format!("{server:?}"), "StreamServer { services: 1 }");
117        let service = server.create().await.unwrap();
118        let debug = format!("{service:?}");
119        assert!(debug.starts_with("StreamService { tokens: {Token(7): (0, "));
120        assert!(debug.contains("conns: Counter("));
121        assert!(debug.contains("count: Cell { value: 0 }"));
122    }
123
124    #[test]
125    fn token_next() {
126        let mut token = Token(0);
127        assert_eq!(token.next(), Token(0));
128        assert_eq!(token.next(), Token(1));
129        assert_eq!(token, Token(2));
130    }
131}
ntex_server/net/mod.rs

ntex_server/net/
mod.rs
