ntex_server/
server.rs

1#![allow(clippy::missing_panics_doc)]
2use std::sync::{Arc, atomic::AtomicBool, atomic::Ordering};
3use std::task::{Context, Poll, ready};
4use std::{future::Future, io, pin::Pin};
5
6use async_channel::Sender;
7use ntex_rt::signals::Signal;
8
9use crate::manager::ServerCommand;
10
11#[derive(Debug)]
12pub(crate) struct ServerShared {
13    pub(crate) paused: AtomicBool,
14}
15
16/// Server controller
17#[derive(Debug)]
18pub struct Server<T> {
19    shared: Arc<ServerShared>,
20    cmd: Sender<ServerCommand<T>>,
21    stop: Option<oneshot::AsyncReceiver<()>>,
22}
23
24impl<T> Server<T> {
25    pub(crate) fn new(cmd: Sender<ServerCommand<T>>, shared: Arc<ServerShared>) -> Self {
26        Server {
27            cmd,
28            shared,
29            stop: None,
30        }
31    }
32
33    /// Start streaming server building process
34    pub fn builder() -> crate::net::ServerBuilder {
35        crate::net::ServerBuilder::default()
36    }
37
38    pub(crate) fn signal(&self, sig: Signal) {
39        let _ = self.cmd.try_send(ServerCommand::Signal(sig));
40    }
41
42    /// Send item to worker pool
43    pub fn process(&self, item: T) -> Result<(), T> {
44        if self.shared.paused.load(Ordering::Acquire) {
45            Err(item)
46        } else if let Err(e) = self.cmd.try_send(ServerCommand::Item(item)) {
47            if let ServerCommand::Item(item) = e.into_inner() {
48                Err(item)
49            } else {
50                panic!()
51            }
52        } else {
53            Ok(())
54        }
55    }
56
57    /// Pause accepting incoming connections
58    ///
59    /// If socket contains some pending connection, they might be dropped.
60    /// All opened connection remains active.
61    pub fn pause(&self) -> impl Future<Output = ()> + use<T> {
62        let (tx, rx) = oneshot::channel();
63        let _ = self.cmd.try_send(ServerCommand::Pause(tx));
64        async move {
65            let _ = rx.await;
66        }
67    }
68
69    /// Resume accepting incoming connections
70    pub fn resume(&self) -> impl Future<Output = ()> + use<T> {
71        let (tx, rx) = oneshot::channel();
72        let _ = self.cmd.try_send(ServerCommand::Resume(tx));
73        async move {
74            let _ = rx.await;
75        }
76    }
77
78    /// Stop incoming connection processing, stop all workers and exit.
79    ///
80    /// If server starts with `spawn()` method, then spawned thread get terminated.
81    pub fn stop(&self, graceful: bool) -> impl Future<Output = ()> + use<T> {
82        let (tx, rx) = oneshot::channel();
83        let _ = self.cmd.try_send(ServerCommand::Stop {
84            graceful,
85            completion: Some(tx),
86        });
87        async move {
88            let _ = rx.await;
89        }
90    }
91}
92
93impl<T> Clone for Server<T> {
94    fn clone(&self) -> Self {
95        Self {
96            cmd: self.cmd.clone(),
97            shared: self.shared.clone(),
98            stop: None,
99        }
100    }
101}
102
103impl<T> Future for Server<T> {
104    type Output = io::Result<()>;
105
106    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
107        let this = self.get_mut();
108
109        if this.stop.is_none() {
110            let (tx, rx) = oneshot::async_channel();
111            if this.cmd.try_send(ServerCommand::NotifyStopped(tx)).is_err() {
112                return Poll::Ready(Ok(()));
113            }
114            this.stop = Some(rx);
115        }
116
117        let _ = ready!(Pin::new(this.stop.as_mut().unwrap()).poll(cx));
118
119        Poll::Ready(Ok(()))
120    }
121}
ntex_server/server.rs

ntex_server/
server.rs
