ntex_server/net/
config.rs

1use std::{cell::RefCell, fmt, io, marker::PhantomData, mem, net, rc::Rc, sync::Arc};
2
3use ntex_io::Io;
4use ntex_service::{IntoService, Pipeline, Service, cfg::SharedCfg};
5use ntex_util::{HashMap, future::BoxFuture};
6
7use super::factory::{FactoryService, FactoryServiceType, NetService, ServerService};
8use super::{ServerAppConfig, Token, builder::bind_addr, socket::Listener};
9
10/// Listener and worker configuration used by [`ServerBuilder::configure`].
11///
12/// Listeners are registered by name. Each worker then runs the
13/// [`on_worker_start`](Self::on_worker_start) callbacks, which attach a
14/// service to each name through [`ServiceRuntime::service`].
15///
16/// [`ServerBuilder::configure`]: super::ServerBuilder::configure
17#[derive(Debug)]
18pub struct ServiceConfig<Cfg: ServerAppConfig>(pub(super) Rc<RefCell<ServiceConfigInner<Cfg>>>);
19
20#[derive(Debug)]
21struct Socket {
22    name: String,
23    sockets: Vec<(Token, Listener, SharedCfg)>,
24}
25
26pub(super) struct ServiceConfigInner<Cfg: ServerAppConfig> {
27    token: Token,
28    on_start_set: bool,
29    on_start: Vec<Box<dyn OnWorkerStart<Cfg::State>>>,
30    sockets: Vec<Socket>,
31    backlog: i32,
32}
33
34impl<Cfg: ServerAppConfig> fmt::Debug for ServiceConfigInner<Cfg> {
35    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
36        f.debug_struct("ServiceConfigInner")
37            .field("token", &self.token)
38            .field("backlog", &self.backlog)
39            .field("sockets", &self.sockets)
40            .finish()
41    }
42}
43
44impl<Cfg: ServerAppConfig> Clone for ServiceConfig<Cfg> {
45    fn clone(&self) -> Self {
46        Self(self.0.clone())
47    }
48}
49
50impl<Cfg: ServerAppConfig> ServiceConfig<Cfg> {
51    pub(super) fn new(token: Token, backlog: i32) -> Self {
52        ServiceConfig(Rc::new(RefCell::new(ServiceConfigInner {
53            token,
54            backlog,
55            sockets: Vec::new(),
56            on_start_set: false,
57            on_start: vec![on_worker_start(async |_| {
58                not_configured();
59                Ok(())
60            })],
61        })))
62    }
63
64    /// Binds TCP listeners under the specified name.
65    ///
66    /// A listener is created for every address resolved from `addr`, using
67    /// the builder's backlog. Binding succeeds if at least one address binds.
68    /// The service for `name` is attached later with
69    /// [`ServiceRuntime::service`].
70    pub fn bind(&self, name: impl AsRef<str>, addr: impl net::ToSocketAddrs) -> io::Result<&Self> {
71        let mut inner = self.0.borrow_mut();
72
73        let sockets = bind_addr(addr, inner.backlog)?;
74        let socket = Socket {
75            name: name.as_ref().to_string(),
76            sockets: sockets
77                .into_iter()
78                .map(|lst| {
79                    (
80                        inner.token.next(),
81                        Listener::from_tcp(lst),
82                        SharedCfg::default(),
83                    )
84                })
85                .collect(),
86        };
87        inner.sockets.push(socket);
88
89        Ok(self)
90    }
91
92    /// Registers an existing TCP listener under the specified name.
93    ///
94    /// The listener is switched to non-blocking mode. The service for `name`
95    /// is attached later with [`ServiceRuntime::service`].
96    pub fn listen(&self, name: impl AsRef<str>, lst: net::TcpListener) -> &Self {
97        let mut inner = self.0.borrow_mut();
98        let socket = Socket {
99            name: name.as_ref().to_string(),
100            sockets: vec![(
101                inner.token.next(),
102                Listener::from_tcp(lst),
103                SharedCfg::default(),
104            )],
105        };
106        inner.sockets.push(socket);
107
108        self
109    }
110
111    /// Registers an asynchronous worker configuration callback.
112    ///
113    /// The callback runs on each worker thread while the worker creates its
114    /// services, and it should attach a service to every registered name.
115    /// Multiple callbacks run in registration order. An error fails the
116    /// worker's service creation. Names left without a service are logged
117    /// as errors, and their connections are dropped.
118    pub fn on_worker_start<F>(&self, f: F) -> &Self
119    where
120        F: AsyncFnOnce(ServiceRuntime<Cfg::State>) -> io::Result<()> + Send + Clone + 'static,
121    {
122        let mut inner = self.0.borrow_mut();
123        if !inner.on_start_set {
124            inner.on_start.clear();
125            inner.on_start_set = true;
126        }
127        inner.on_start.push(on_worker_start(f));
128        self
129    }
130
131    pub(super) fn into_factory(
132        self,
133    ) -> (
134        Token,
135        Vec<(Token, String, Listener)>,
136        FactoryServiceType<Cfg>,
137    ) {
138        let mut inner = self.0.borrow_mut();
139
140        let mut sockets = Vec::new();
141        let mut names = HashMap::default();
142        for (idx, s) in mem::take(&mut inner.sockets).into_iter().enumerate() {
143            names.insert(
144                s.name.clone(),
145                Entry {
146                    idx,
147                    name: s.name.clone(),
148                    tokens: s
149                        .sockets
150                        .iter()
151                        .map(|(token, _, cfg)| (*token, cfg.clone()))
152                        .collect(),
153                },
154            );
155
156            sockets.extend(
157                s.sockets
158                    .into_iter()
159                    .map(|(token, lst, _)| (token, s.name.clone(), lst)),
160            );
161        }
162
163        (
164            inner.token,
165            sockets,
166            Box::new(ConfiguredService {
167                names,
168                on_start: mem::take(&mut inner.on_start),
169            }),
170        )
171    }
172}
173
174/// Per-worker runtime used to attach services to named listeners.
175///
176/// Passed to [`ServiceConfig::on_worker_start`] callbacks.
177pub struct ServiceRuntime<Cfg>(Cfg, Rc<RefCell<ServiceRuntimeInner>>);
178
179#[derive(Debug, Clone)]
180struct Entry {
181    idx: usize,
182    name: String,
183    tokens: Vec<(Token, SharedCfg)>,
184}
185
186struct ServiceRuntimeInner {
187    names: HashMap<String, Entry>,
188    services: Vec<Option<Box<dyn NetService>>>,
189}
190
191impl<Cfg> fmt::Debug for ServiceRuntime<Cfg> {
192    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
193        let inner = self.1.borrow();
194        f.debug_struct("ServiceRuntimer")
195            .field("names", &inner.names)
196            .finish()
197    }
198}
199
200impl<Cfg: Clone + 'static> ServiceRuntime<Cfg> {
201    fn new(st: Cfg, names: HashMap<String, Entry>) -> Self {
202        let services = (0..names.len()).map(|_| None).collect();
203        ServiceRuntime(
204            st,
205            Rc::new(RefCell::new(ServiceRuntimeInner { names, services })),
206        )
207    }
208
209    fn validate(&self) {
210        let inner = self.1.as_ref().borrow();
211        for (name, item) in &inner.names {
212            if inner.services[item.idx].is_none() {
213                log::error!("Service {name:?} is not configured");
214            }
215        }
216    }
217
218    /// Returns the application state passed to services.
219    pub fn cfg(&self) -> &Cfg {
220        &self.0
221    }
222
223    /// Attaches a service to the listeners registered under `name`.
224    ///
225    /// The name must be registered during the configuration stage with
226    /// [`ServiceConfig::bind`] or [`ServiceConfig::listen`]. `cfg` is the I/O
227    /// configuration for connections accepted by those listeners. Attaching
228    /// a service to the same name again replaces the previous one.
229    ///
230    /// # Panics
231    ///
232    /// Panics if no listener is registered under `name`.
233    pub fn service<S>(
234        &self,
235        name: &str,
236        cfg: impl Into<SharedCfg>,
237        svc: impl IntoService<S, Cfg, Io>,
238    ) -> &Self
239    where
240        S: Service<Cfg, Io> + 'static,
241    {
242        let shared = cfg.into();
243        let mut inner = self.1.borrow_mut();
244        if let Some(entry) = inner.names.get_mut(name) {
245            let idx = entry.idx;
246            for token in &mut entry.tokens {
247                token.1 = shared.clone();
248            }
249            let pipeline = Pipeline::new(
250                self.0.clone(),
251                svc.into_service().map(|_| ()).map_err(|_| ()),
252            );
253            let svc: Box<dyn NetService> = Box::new(ServerService { pipeline });
254            inner.services[idx] = Some(svc);
255        } else {
256            panic!("Unknown service: {name:?}");
257        }
258        self
259    }
260
261    /// Returns a runtime that passes `st` as the state to services.
262    ///
263    /// Services registered through the returned runtime share the same
264    /// listeners as this runtime.
265    pub fn map_cfg<T>(&self, st: T) -> ServiceRuntime<T>
266    where
267        T: Clone + 'static,
268    {
269        ServiceRuntime(st, self.1.clone())
270    }
271}
272
273struct ConfiguredService<Cfg: ServerAppConfig> {
274    names: HashMap<String, Entry>,
275    on_start: Vec<Box<dyn OnWorkerStart<Cfg::State>>>,
276}
277
278impl<Cfg: ServerAppConfig> FactoryService<Cfg> for ConfiguredService<Cfg> {
279    fn clo(&self) -> FactoryServiceType<Cfg> {
280        Box::new(Self {
281            names: self.names.clone(),
282            on_start: self.on_start.iter().map(|cb| (*cb).clo()).collect(),
283        })
284    }
285
286    fn create(
287        &self,
288        st: Cfg::State,
289    ) -> BoxFuture<'static, io::Result<Vec<(Box<dyn NetService>, Arc<str>, Vec<(Token, SharedCfg)>)>>>
290    {
291        // configure services
292        let rt = ServiceRuntime::new(st.clone(), self.names.clone());
293        let on_start: Vec<_> = self
294            .on_start
295            .iter()
296            .map(|cb| cb.run(ServiceRuntime(st.clone(), rt.1.clone())))
297            .collect();
298
299        // construct services
300        Box::pin(async move {
301            for fut in on_start {
302                fut.await?;
303            }
304            rt.validate();
305
306            let names = mem::take(&mut rt.1.borrow_mut().names);
307            let mut services = mem::take(&mut rt.1.borrow_mut().services);
308
309            let mut res = Vec::new();
310            while let Some(svc) = services.pop() {
311                if let Some(svc) = svc {
312                    for entry in names.values() {
313                        if entry.idx == services.len() {
314                            res.push((
315                                svc,
316                                std::sync::Arc::from(entry.name.clone()),
317                                entry.tokens.clone(),
318                            ));
319                            break;
320                        }
321                    }
322                }
323            }
324            Ok(res)
325        })
326    }
327}
328
329fn not_configured() {
330    log::error!("Service is not configured");
331}
332
333trait OnWorkerStart<Cfg>: Send {
334    fn clo(&self) -> Box<dyn OnWorkerStart<Cfg>>;
335
336    fn run(&self, rt: ServiceRuntime<Cfg>) -> BoxFuture<'static, io::Result<()>>;
337}
338
339struct OnWorkerStartImpl<F, Cfg> {
340    f: F,
341    st: PhantomData<Cfg>,
342}
343
344fn on_worker_start<F, Cfg>(f: F) -> Box<dyn OnWorkerStart<Cfg> + Send>
345where
346    F: AsyncFnOnce(ServiceRuntime<Cfg>) -> io::Result<()> + Send + Clone + 'static,
347    Cfg: 'static,
348{
349    Box::new(OnWorkerStartImpl { f, st: PhantomData })
350}
351
352impl<F, Cfg> OnWorkerStart<Cfg> for OnWorkerStartImpl<F, Cfg>
353where
354    F: AsyncFnOnce(ServiceRuntime<Cfg>) -> io::Result<()> + Send + Clone + 'static,
355    Cfg: 'static,
356{
357    fn clo(&self) -> Box<dyn OnWorkerStart<Cfg>> {
358        Box::new(Self {
359            f: self.f.clone(),
360            st: PhantomData,
361        })
362    }
363
364    fn run(&self, rt: ServiceRuntime<Cfg>) -> BoxFuture<'static, io::Result<()>> {
365        let f = self.f.clone();
366        Box::pin(async move { (f)(rt).await })
367    }
368}
369
370// SAFETY: Send cannot be provided authomatically because of R param
371// but R always get executed in one thread and never leave it
372unsafe impl<F, Cfg> Send for OnWorkerStartImpl<F, Cfg> where F: Send {}
ntex_server/net/config.rs

ntex_server/net/
config.rs
