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

ntex_server/net/
service.rs
