futu_server/
listener_status.rs

1//! Bind-success facts shared by daemon surface listeners.
2//!
3//! A surface reports `Opened` only after its socket bind succeeds. The event
4//! contains a stable surface token plus opaque in-process publication and
5//! active acknowledgements; no bind address, credential, request, response,
6//! account, or backend endpoint is carried across this boundary.
7
8use std::collections::HashSet;
9use std::sync::Arc;
10
11use parking_lot::Mutex;
12use tokio::sync::mpsc;
13use tokio::sync::{oneshot, watch};
14
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
16pub enum ListenerSurface {
17    Ftapi,
18    WebSocket,
19    Rest,
20    Grpc,
21    Telnet,
22}
23
24impl ListenerSurface {
25    pub const fn as_str(self) -> &'static str {
26        match self {
27            Self::Ftapi => "ftapi",
28            Self::WebSocket => "websocket",
29            Self::Rest => "rest",
30            Self::Grpc => "grpc",
31            Self::Telnet => "telnet",
32        }
33    }
34}
35
36#[derive(Debug)]
37pub enum ListenerBindEvent {
38    Opened {
39        surface: ListenerSurface,
40        publish: oneshot::Sender<Arc<ServingRegistry>>,
41        active: oneshot::Receiver<()>,
42    },
43    Failed(ListenerSurface),
44}
45
46impl PartialEq for ListenerBindEvent {
47    fn eq(&self, other: &Self) -> bool {
48        match (self, other) {
49            (Self::Opened { surface: left, .. }, Self::Opened { surface: right, .. }) => {
50                left == right
51            }
52            (Self::Failed(left), Self::Failed(right)) => left == right,
53            _ => false,
54        }
55    }
56}
57
58impl Eq for ListenerBindEvent {}
59
60pub type ListenerBindEventSender = mpsc::UnboundedSender<ListenerBindEvent>;
61
62#[derive(Debug, Default)]
63pub struct ServingRegistry {
64    active: Mutex<HashSet<ListenerSurface>>,
65}
66
67impl ServingRegistry {
68    pub fn activate(
69        self: &Arc<Self>,
70        surface: ListenerSurface,
71    ) -> std::io::Result<ActiveListenerGuard> {
72        if !self.active.lock().insert(surface) {
73            return Err(std::io::Error::new(
74                std::io::ErrorKind::AlreadyExists,
75                format!("{} listener is already active", surface.as_str()),
76            ));
77        }
78        Ok(ActiveListenerGuard {
79            registry: Arc::clone(self),
80            surface,
81        })
82    }
83
84    pub fn with_all_active<T>(
85        &self,
86        expected: &[ListenerSurface],
87        on_active: impl FnOnce() -> T,
88    ) -> std::io::Result<T> {
89        let active = self.active.lock();
90        if active.len() != expected.len()
91            || expected.iter().any(|surface| !active.contains(surface))
92        {
93            return Err(std::io::Error::new(
94                std::io::ErrorKind::NotConnected,
95                "not every enabled listener is active at readiness publication",
96            ));
97        }
98        Ok(on_active())
99    }
100}
101
102#[derive(Debug)]
103pub struct ActiveListenerGuard {
104    registry: Arc<ServingRegistry>,
105    surface: ListenerSurface,
106}
107
108impl Drop for ActiveListenerGuard {
109    fn drop(&mut self) {
110        self.registry.active.lock().remove(&self.surface);
111    }
112}
113
114pub async fn notify_listener_opened(
115    events: &Option<ListenerBindEventSender>,
116    surface: ListenerSurface,
117    shutdown_rx: &watch::Receiver<bool>,
118) -> std::io::Result<Option<ActiveListenerGuard>> {
119    let Some(events) = events else {
120        return Ok(None);
121    };
122    if *shutdown_rx.borrow() {
123        notify_listener_failed(&Some(events.clone()), surface);
124        return Err(std::io::Error::new(
125            std::io::ErrorKind::Interrupted,
126            format!(
127                "{} listener shutdown was already requested before startup publication",
128                surface.as_str()
129            ),
130        ));
131    }
132
133    let (publish, published) = oneshot::channel();
134    let (active, activated) = oneshot::channel();
135    events
136        .send(ListenerBindEvent::Opened {
137            surface,
138            publish,
139            active: activated,
140        })
141        .map_err(|_| {
142            std::io::Error::new(
143                std::io::ErrorKind::BrokenPipe,
144                format!(
145                    "{} listener startup coordinator dropped before publication",
146                    surface.as_str()
147                ),
148            )
149        })?;
150    let registry = published.await.map_err(|_| {
151        std::io::Error::new(
152            std::io::ErrorKind::Interrupted,
153            format!(
154                "{} listener startup publication was cancelled",
155                surface.as_str()
156            ),
157        )
158    })?;
159    if *shutdown_rx.borrow() {
160        notify_listener_failed(&Some(events.clone()), surface);
161        return Err(std::io::Error::new(
162            std::io::ErrorKind::Interrupted,
163            format!(
164                "{} listener shutdown was requested before activation",
165                surface.as_str()
166            ),
167        ));
168    }
169    let guard = registry.activate(surface)?;
170    active.send(()).map_err(|_| {
171        std::io::Error::new(
172            std::io::ErrorKind::Interrupted,
173            format!(
174                "{} listener active acknowledgement was cancelled",
175                surface.as_str()
176            ),
177        )
178    })?;
179    Ok(Some(guard))
180}
181
182pub fn notify_listener_failed(events: &Option<ListenerBindEventSender>, surface: ListenerSurface) {
183    if let Some(events) = events {
184        let _ = events.send(ListenerBindEvent::Failed(surface));
185    }
186}
futu_server/listener_status.rs

futu_server/
listener_status.rs
