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

ntex_server/net/
service.rs
