ntex_server/net/
service.rs

1use std::{fmt, io, sync::Arc};
2
3use ntex_service::{Ctx, Service, cfg::SharedCfg};
4use ntex_util::{HashMap, future::join_all, services::Counter};
5
6use crate::{ServerAppConfig, ServerConfiguration};
7
8use super::accept::{AcceptNotify, AcceptorCommand};
9use super::factory::{FactoryServiceType, NetService};
10use super::{MAX_CONNS_COUNTER, Token, socket::Connection};
11
12/// Worker configuration for network servers.
13///
14/// Created by [`ServerBuilder::run`](super::ServerBuilder::run). Each worker
15/// creates its application state and the services for all registered
16/// listeners.
17pub struct StreamServer<Cfg> {
18    accept: AcceptNotify,
19    state: Arc<Cfg>,
20    services: Vec<FactoryServiceType<Cfg>>,
21}
22
23impl<Cfg: ServerAppConfig> StreamServer<Cfg> {
24    pub(crate) fn new(
25        accept: AcceptNotify,
26        state: Arc<Cfg>,
27        services: Vec<FactoryServiceType<Cfg>>,
28    ) -> Self {
29        Self {
30            accept,
31            state,
32            services,
33        }
34    }
35}
36
37impl<Cfg> fmt::Debug for StreamServer<Cfg> {
38    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
39        f.debug_struct("StreamServer")
40            .field("services", &self.services.len())
41            .finish()
42    }
43}
44
45/// Worker service factory.
46impl<Cfg: ServerAppConfig> ServerConfiguration for StreamServer<Cfg> {
47    type Item = Connection;
48    type Service = StreamService;
49
50    /// Create service for handling connections
51    async fn create(&self) -> io::Result<Self::Service> {
52        // construct configuration
53        let cfg = self.state.create().await?;
54
55        // construct services
56        let mut tokens = HashMap::default();
57        let mut services = Vec::new();
58
59        for info in &self.services {
60            for (svc, _, svc_tokens) in info.create(cfg.clone()).await.map_err(io::Error::other)? {
61                services.push(svc);
62                let idx = services.len() - 1;
63                for (token, cfg) in &svc_tokens {
64                    tokens.insert(*token, (idx, cfg.clone()));
65                }
66            }
67        }
68
69        Ok(StreamService {
70            services,
71            tokens,
72            conns: MAX_CONNS_COUNTER.with(Clone::clone),
73        })
74    }
75
76    /// Pause the server.
77    fn pause(&self) {
78        self.accept.send(AcceptorCommand::Pause);
79    }
80
81    /// Resume the server.
82    fn resume(&self) {
83        self.accept.send(AcceptorCommand::Resume);
84    }
85
86    /// Terminate the server.
87    fn terminate(&self) {
88        self.accept.send(AcceptorCommand::Terminate);
89    }
90
91    /// Stop the server.
92    async fn stop(&self) {
93        let (tx, rx) = oneshot::channel();
94        self.accept.send(AcceptorCommand::Stop(tx));
95        let _ = rx.await;
96    }
97}
98
99impl<Cfg: ServerAppConfig> Clone for StreamServer<Cfg> {
100    fn clone(&self) -> Self {
101        Self {
102            state: self.state.clone(),
103            accept: self.accept.clone(),
104            services: self.services.iter().map(|s| s.clo()).collect(),
105        }
106    }
107}
108
109/// Per-worker service that dispatches connections to listener services.
110pub struct StreamService {
111    tokens: HashMap<Token, (usize, SharedCfg)>,
112    services: Vec<Box<dyn NetService>>,
113    conns: Counter,
114}
115
116impl fmt::Debug for StreamService {
117    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
118        f.debug_struct("StreamService")
119            .field("tokens", &self.tokens)
120            .field("conns", &self.conns)
121            .finish()
122    }
123}
124
125impl Service<(), Connection> for StreamService {
126    type Res = ();
127    type Error = ();
128
129    async fn ready(&self, _: Ctx<'_, Self, ()>) -> Result<(), Self::Error> {
130        if !self.conns.is_available() {
131            self.conns.available().await;
132        }
133        for (idx, svc) in self.services.iter().enumerate() {
134            if svc.ready().await.is_err() {
135                for (idx_, cfg) in self.tokens.values() {
136                    if idx == *idx_ {
137                        log::error!("{}: Service readiness has failed", cfg.tag());
138                        break;
139                    }
140                }
141                return Err(());
142            }
143        }
144
145        Ok(())
146    }
147
148    async fn shutdown(&self, _: Ctx<'_, Self, ()>) {
149        let _ = join_all(self.services.iter().map(|s| s.shutdown())).await;
150        log::info!(
151            "Worker service shutdown, {} connections",
152            super::num_connections()
153        );
154    }
155
156    #[allow(clippy::unused_async_trait_impl)]
157    async fn call(&self, con: Connection, _: Ctx<'_, Self, ()>) -> Result<(), ()> {
158        if let Some((idx, cfg)) = self.tokens.get(&con.token) {
159            let stream = con.io.convert(cfg.clone()).map_err(|e| {
160                log::error!("Cannot convert to an async io stream: {e}");
161            })?;
162
163            self.services[*idx].call(stream, self.conns.get());
164            Ok(())
165        } else {
166            log::error!("Cannot get handler service for connection: {con:?}");
167            Err(())
168        }
169    }
170}
ntex_server/net/service.rs

ntex_server/net/
service.rs
