zerodds_webdds_bridge/daemon/
server.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 ZeroDDS Contributors
3
4//! The `zerodds-webdds-bridged` accept loop.
5//!
6//! Sync, blocking I/O — the workspace convention (`crates/monitor`,
7//! `crates/coap-bridge`, `crates/websocket-bridge`). One thread per TCP
8//! connection, looping `http1::read_request`/`write_response` for
9//! keep-alive; the request's path decides REST (`/dds/rest1/...`), SOAP
10//! (`/soap/...`) or WSDL (`/wsdl`) dispatch, all against the one shared
11//! [`Gateway`]. `metrics.enabled` also starts the admin `/catalog`+
12//! `/healthz`+`/metrics` endpoint on its own port
13//! ([`runtime_common::serve_admin_endpoints`]) — the daemon family's
14//! own convention, not a DDS-WEB 1.0 requirement.
15
16// Daemon logging: `eprintln!` is the sink until a workspace tracing
17// stack is wired in — the same allowance every other
18// `zerodds-*-bridged` daemon's `runtime_common`/`server` module takes.
19#![allow(clippy::print_stderr)]
20
21use std::io::BufReader;
22use std::net::{TcpListener, TcpStream};
23use std::path::PathBuf;
24use std::sync::atomic::{AtomicBool, Ordering};
25use std::sync::{Arc, Mutex, RwLock};
26use std::thread::{self, JoinHandle};
27use std::time::Duration;
28
29use crate::gateway::Gateway;
30use crate::http1::{self, Response};
31use crate::metadata::MetadataStore;
32use crate::rpc_dispatch::RpcDispatcher;
33use crate::{rest_service, soap_service};
34
35use super::config::DaemonConfig;
36use super::runtime_common;
37use super::security::{self, SecurityCtx, serve_tls_handshake};
38use zerodds_bridge_security::rustls;
39
40/// Admin endpoint address when `metrics.enabled` is set without
41/// `metrics.listen`.
42pub const DEFAULT_METRICS_LISTEN: &str = "127.0.0.1:9095";
43
44/// Top-level error of the daemon.
45#[derive(Debug)]
46pub enum ServerError {
47    /// A setting that cannot be served.
48    Config(String),
49    /// TLS material that does not load.
50    Tls(String),
51    /// Bind error.
52    Bind(String),
53    /// DCPS init error.
54    Dds(String),
55    /// I/O error during operation.
56    Io(String),
57}
58
59impl core::fmt::Display for ServerError {
60    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
61        match self {
62            Self::Config(m) => write!(f, "config error: {m}"),
63            Self::Tls(m) => write!(f, "tls error: {m}"),
64            Self::Bind(m) => write!(f, "bind error: {m}"),
65            Self::Dds(m) => write!(f, "dds error: {m}"),
66            Self::Io(m) => write!(f, "io error: {m}"),
67        }
68    }
69}
70
71impl std::error::Error for ServerError {}
72
73/// A config reload hook, invoked on SIGHUP.
74pub type ConfigLoader = Box<dyn Fn() -> Result<DaemonConfig, String> + Send + Sync>;
75
76/// Daemon handle. On drop, no implicit shutdown — call
77/// [`DaemonHandle::shutdown`] explicitly (mirrors every other
78/// `zerodds-*-bridged` daemon).
79pub struct DaemonHandle {
80    stop: Arc<AtomicBool>,
81    accept_thread: Option<JoinHandle<()>>,
82    /// Bound HTTP address (may differ from the configured listen if
83    /// port 0 was asked for).
84    pub local_addr: String,
85    /// Bound admin (`/metrics`) address, if `metrics.enabled`.
86    pub admin_addr: Option<String>,
87    /// SIGHUP sets this.
88    pub reload_flag: Arc<AtomicBool>,
89    /// `true` once the listener is up.
90    pub healthy: Arc<AtomicBool>,
91    /// The shared gateway — exported for tests that want to inspect or
92    /// drive it directly without going through HTTP.
93    pub gateway: Arc<Mutex<Gateway>>,
94}
95
96impl DaemonHandle {
97    /// Whether SIGTERM/SIGINT (or [`Self::shutdown`]) asked the daemon
98    /// to stop.
99    #[must_use]
100    pub fn stop_requested(&self) -> bool {
101        self.stop.load(Ordering::SeqCst)
102    }
103
104    /// Initiates a graceful shutdown: stops accepting new connections
105    /// and joins the accept thread. Open connections are not force-
106    /// closed — they finish their current request/response and then
107    /// see `stop` on their next keep-alive iteration.
108    pub fn shutdown(&mut self) {
109        self.stop.store(true, Ordering::SeqCst);
110        self.healthy.store(false, Ordering::SeqCst);
111        // Unblock `accept()` by connecting to ourselves once.
112        if let Ok(mut s) = TcpStream::connect(&self.local_addr) {
113            use std::io::Write;
114            let _ = s.write_all(b"GET / HTTP/1.1\r\nConnection: close\r\n\r\n");
115        }
116        if let Some(h) = self.accept_thread.take() {
117            let _ = h.join();
118        }
119    }
120}
121
122/// Starts the daemon. Blocking-thread accept loop, returned already
123/// running.
124///
125/// # Errors
126/// See [`ServerError`].
127pub fn start(cfg: DaemonConfig) -> Result<DaemonHandle, ServerError> {
128    start_with_reload(cfg, None)
129}
130
131/// As [`start`], with an optional SIGHUP reload hook.
132///
133/// # Errors
134/// See [`ServerError`].
135pub fn start_with_reload(
136    cfg: DaemonConfig,
137    reload: Option<ConfigLoader>,
138) -> Result<DaemonHandle, ServerError> {
139    let type_sources: std::collections::BTreeMap<String, crate::gateway::TypeSource> = cfg
140        .types
141        .iter()
142        .filter_map(|t| {
143            let source = if let Some(idl) = &t.idl_file {
144                crate::gateway::TypeSource::Idl(PathBuf::from(idl))
145            } else if let Some(xml) = &t.xml_file {
146                crate::gateway::TypeSource::Xml(PathBuf::from(xml))
147            } else {
148                return None;
149            };
150            Some((t.type_name.clone(), source))
151        })
152        .collect();
153    let metadata = Arc::new(MetadataStore::new());
154    let gateway = Arc::new(Mutex::new(Gateway::new(
155        type_sources,
156        Arc::clone(&metadata),
157    )));
158
159    let (sec_ctx, rotating) =
160        security::ctx_from_daemon_config(&cfg).map_err(|e| ServerError::Tls(e.to_string()))?;
161    let sec_ctx = Arc::new(RwLock::new(sec_ctx));
162    let rotating = rotating.map(Arc::new);
163
164    let wsdl_xml = Arc::new(build_wsdl(&cfg, &cfg.listen).map_err(ServerError::Config)?);
165    let rpc_dispatchers = Arc::new(build_rpc_dispatchers(&cfg)?);
166
167    let listener = TcpListener::bind(&cfg.listen).map_err(|e| ServerError::Bind(e.to_string()))?;
168    let local_addr = listener
169        .local_addr()
170        .map_err(|e| ServerError::Bind(e.to_string()))?
171        .to_string();
172
173    let stop = Arc::new(AtomicBool::new(false));
174    let reload_flag = Arc::new(AtomicBool::new(false));
175    let healthy = Arc::new(AtomicBool::new(true));
176    install_signal_watcher(Arc::clone(&stop), Arc::clone(&reload_flag))
177        .map_err(|e| ServerError::Io(e.to_string()))?;
178
179    // Admin endpoint (`/catalog`+`/healthz`+`/metrics`) — the daemon
180    // family's own convention, not a DDS-WEB 1.0 requirement; see
181    // `daemon::runtime_common`'s module doc.
182    let registry = Arc::new(zerodds_monitor::Registry::new());
183    let admin_addr = if cfg.metrics_enabled {
184        let addr_str = if cfg.metrics_addr.is_empty() {
185            DEFAULT_METRICS_LISTEN
186        } else {
187            &cfg.metrics_addr
188        };
189        let addr = addr_str
190            .parse()
191            .map_err(|_| ServerError::Config(format!("bad metrics.listen: {addr_str}")))?;
192        let catalog = Arc::new(runtime_common::CatalogSnapshot::from_config(&cfg));
193        runtime_common::serve_admin_endpoints(
194            addr,
195            catalog,
196            Arc::clone(&registry),
197            Arc::clone(&healthy),
198            Arc::clone(&stop),
199        )
200        .map_err(|e| ServerError::Bind(e.to_string()))?;
201        Some(addr_str.to_string())
202    } else {
203        None
204    };
205
206    // SIGHUP: re-reads the config and swaps in the new auth/ACL (and, via
207    // `rotating`, the new TLS cert material) — matching every other
208    // `zerodds-*-bridged` daemon's documented SIGHUP contract. A config
209    // that fails to load or to validate changes nothing and is logged.
210    if let Some(load) = reload {
211        let reload_flag_watch = Arc::clone(&reload_flag);
212        let stop_watch = Arc::clone(&stop);
213        let sec_ctx_watch = Arc::clone(&sec_ctx);
214        let rotating_watch = rotating.clone();
215        let _ = thread::Builder::new()
216            .name("zerodds-webdds-reload".into())
217            .spawn(move || {
218                while !stop_watch.load(Ordering::SeqCst) {
219                    if reload_flag_watch.swap(false, Ordering::SeqCst) {
220                        match load() {
221                            Ok(new_cfg) => match security::ctx_from_daemon_config(&new_cfg) {
222                                Ok((new_ctx, _new_rot)) => {
223                                    if let Ok(mut w) = sec_ctx_watch.write() {
224                                        *w = new_ctx;
225                                    }
226                                    // Cert/key/CA *paths* are fixed for the
227                                    // process's lifetime (an `Arc<RotatingTlsConfig>`
228                                    // is shared with every already-spawned
229                                    // connection thread); a config that keeps
230                                    // them and only renewed the file content
231                                    // is what `reload()` re-reads. A config
232                                    // that names *different* paths needs a
233                                    // restart — logged, not applied silently.
234                                    if let Some(rot) = rotating_watch.as_ref() {
235                                        if let Err(e) = rot.reload() {
236                                            eprintln!(
237                                                "[zerodds-webdds-bridged] tls reload refused: {e}"
238                                            );
239                                        }
240                                    }
241                                }
242                                Err(e) => eprintln!("[zerodds-webdds-bridged] reload refused: {e}"),
243                            },
244                            Err(e) => eprintln!("[zerodds-webdds-bridged] reload refused: {e}"),
245                        }
246                    }
247                    thread::sleep(Duration::from_millis(200));
248                }
249            });
250    }
251
252    let max_body = cfg.max_body_bytes;
253    let stop_for_loop = Arc::clone(&stop);
254    let gateway_for_loop = Arc::clone(&gateway);
255    let accept_thread = thread::Builder::new()
256        .name("zerodds-webdds-accept".into())
257        .spawn(move || {
258            for stream in listener.incoming() {
259                if stop_for_loop.load(Ordering::SeqCst) {
260                    break;
261                }
262                let Ok(tcp) = stream else { continue };
263                let gw = Arc::clone(&gateway_for_loop);
264                let meta = Arc::clone(&metadata);
265                let ctx = Arc::clone(&sec_ctx);
266                let wsdl = Arc::clone(&wsdl_xml);
267                let rpc = Arc::clone(&rpc_dispatchers);
268                let rot = rotating.clone();
269                thread::spawn(move || {
270                    let web_stream = match rot.as_ref() {
271                        Some(rot) => {
272                            match serve_tls_handshake(rot.current(), tcp, Duration::from_secs(5)) {
273                                Ok((tcp, conn, _mtls_subj)) => {
274                                    WebStream::Tls(Box::new(rustls::StreamOwned::new(conn, tcp)))
275                                }
276                                Err(e) => {
277                                    eprintln!("[zerodds-webdds-bridged] tls handshake error: {e}");
278                                    return;
279                                }
280                            }
281                        }
282                        None => WebStream::Plain(tcp),
283                    };
284                    handle_connection(web_stream, &gw, &meta, &ctx, &wsdl, &rpc, max_body);
285                });
286            }
287        })
288        .map_err(|e| ServerError::Io(e.to_string()))?;
289
290    Ok(DaemonHandle {
291        stop,
292        accept_thread: Some(accept_thread),
293        local_addr,
294        admin_addr,
295        reload_flag,
296        healthy,
297        gateway,
298    })
299}
300
301/// Either side of an accepted connection — plain when `tls.enabled` is
302/// false, a server-side rustls stream when it is true. One `BufReader`
303/// wraps it for reads; a write goes straight to the inner stream via
304/// [`BufReader::get_mut`] (which never touches the read buffer), so the
305/// two paths need no `Clone` — `StreamOwned`'s rustls session state
306/// could not provide one anyway.
307enum WebStream {
308    /// `tls.enabled=false`.
309    Plain(TcpStream),
310    /// Server-side TLS stream on an accepted connection.
311    Tls(Box<rustls::StreamOwned<rustls::ServerConnection, TcpStream>>),
312}
313
314impl WebStream {
315    fn set_read_timeout(&mut self, dur: Option<Duration>) -> std::io::Result<()> {
316        match self {
317            Self::Plain(s) => s.set_read_timeout(dur),
318            Self::Tls(s) => s.sock.set_read_timeout(dur),
319        }
320    }
321}
322
323impl std::io::Read for WebStream {
324    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
325        match self {
326            Self::Plain(s) => s.read(buf),
327            Self::Tls(s) => s.read(buf),
328        }
329    }
330}
331
332impl std::io::Write for WebStream {
333    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
334        match self {
335            Self::Plain(s) => s.write(buf),
336            Self::Tls(s) => s.write(buf),
337        }
338    }
339    fn flush(&mut self) -> std::io::Result<()> {
340        match self {
341            Self::Plain(s) => s.flush(),
342            Self::Tls(s) => s.flush(),
343        }
344    }
345}
346
347fn handle_connection(
348    mut stream: WebStream,
349    gateway: &Arc<Mutex<Gateway>>,
350    metadata: &Arc<MetadataStore>,
351    sec_ctx: &Arc<RwLock<SecurityCtx>>,
352    wsdl_xml: &Arc<String>,
353    rpc_dispatchers: &Arc<Vec<RpcDispatcher>>,
354    max_body: usize,
355) {
356    if stream
357        .set_read_timeout(Some(Duration::from_secs(30)))
358        .is_err()
359    {
360        return;
361    }
362    let mut reader = BufReader::new(stream);
363    loop {
364        let req = match http1::read_request(&mut reader, max_body) {
365            Ok(r) => r,
366            Err(http1::HttpError::ConnectionClosed) => break,
367            Err(http1::HttpError::BodyTooLarge) => {
368                let _ = http1::write_response(
369                    reader.get_mut(),
370                    &Response::empty(413, "Payload Too Large"),
371                    false,
372                );
373                break;
374            }
375            Err(_) => {
376                let _ = http1::write_response(
377                    reader.get_mut(),
378                    &Response::empty(400, "Bad Request"),
379                    false,
380                );
381                break;
382            }
383        };
384        let keep_alive = req.keep_alive();
385        let resp = route_http(&req, gateway, metadata, sec_ctx, wsdl_xml, rpc_dispatchers);
386        if http1::write_response(reader.get_mut(), &resp, keep_alive).is_err() {
387            break;
388        }
389        if !keep_alive {
390            break;
391        }
392    }
393}
394
395fn route_http(
396    req: &http1::Request,
397    gateway: &Arc<Mutex<Gateway>>,
398    metadata: &Arc<MetadataStore>,
399    sec_ctx: &Arc<RwLock<SecurityCtx>>,
400    wsdl_xml: &Arc<String>,
401    rpc_dispatchers: &Arc<Vec<RpcDispatcher>>,
402) -> Response {
403    let path = req.path();
404    if path.starts_with("/wsdl") {
405        return Response::new(200, "OK", "text/xml", wsdl_xml.as_bytes().to_vec());
406    }
407
408    let ctx = sec_ctx
409        .read()
410        .unwrap_or_else(std::sync::PoisonError::into_inner);
411    let subject = match security::authenticate_http(&ctx.auth, req.header("authorization"), None) {
412        Ok(s) => s,
413        Err(_) => return Response::empty(401, "Unauthorized"),
414    };
415    let op = security::op_of_method(&req.method);
416    let resource = security::application_of_path(path).unwrap_or_else(|| "*".to_string());
417    let permit = security::authorize(&ctx.acl, &subject, op, &resource);
418    drop(ctx);
419
420    if path.starts_with("/soap") {
421        let body_text = String::from_utf8_lossy(&req.body).to_string();
422        let xml = soap_service::dispatch_envelope(gateway, rpc_dispatchers, &body_text, permit);
423        return Response::new(200, "OK", "application/soap+xml", xml.into_bytes());
424    }
425    rest_service::dispatch(gateway, metadata, req, permit)
426}
427
428/// Builds the SOAP WSDL document: the fixed DDS-WEB object-model
429/// operations plus every `interface`'s operations from `cfg.interfaces`
430/// (`crate::wsdl_bridge`).
431fn build_wsdl(cfg: &DaemonConfig, listen: &str) -> Result<String, String> {
432    let endpoint = format!("http://{listen}/soap");
433    let mut generator = zerodds_soap::wsdl::WsdlGenerator::new(
434        "ZeroDDSWebDDS",
435        "urn:zerodds:webdds",
436        &endpoint,
437        zerodds_soap::wsdl::WsdlVersion::V11,
438    );
439    for op in crate::wsdl_bridge::builtin_operations() {
440        generator = generator.operation(op);
441    }
442    for idl_path in &cfg.interfaces {
443        let source = std::fs::read_to_string(idl_path).map_err(|e| format!("{idl_path}: {e}"))?;
444        let ast = zerodds_wire_probe::encode::model_from_idl(&source, idl_path)
445            .map_err(|e| format!("{idl_path}: {e}"))?;
446        let by_service =
447            crate::wsdl_bridge::operations_of(&ast).map_err(|e| format!("{idl_path}: {e:?}"))?;
448        for (_, ops) in by_service {
449            for op in ops {
450                generator = generator.operation(op);
451            }
452        }
453    }
454    Ok(generator.render())
455}
456
457/// Builds one [`RpcDispatcher`] per `cfg.interfaces` IDL file, each on
458/// its own DDS-RPC client `DomainParticipant` on `cfg.domain` — the
459/// config key the man page has documented since round 1 as "reserved
460/// for a future default"; this is that default's first user.
461///
462/// Empty (no participant created at all) where `cfg.interfaces` is
463/// empty — a daemon with no business interfaces configured opens no
464/// DDS-RPC client participant.
465fn build_rpc_dispatchers(cfg: &DaemonConfig) -> Result<Vec<RpcDispatcher>, ServerError> {
466    if cfg.interfaces.is_empty() {
467        return Ok(Vec::new());
468    }
469    let factory = zerodds_dcps::DomainParticipantFactory::instance();
470    let mut out = Vec::with_capacity(cfg.interfaces.len());
471    for idl_path in &cfg.interfaces {
472        let source = std::fs::read_to_string(idl_path)
473            .map_err(|e| ServerError::Config(format!("{idl_path}: {e}")))?;
474        let ast = zerodds_wire_probe::encode::model_from_idl(&source, idl_path)
475            .map_err(|e| ServerError::Config(format!("{idl_path}: {e}")))?;
476        let participant = factory
477            .create_participant(cfg.domain, zerodds_dcps::DomainParticipantQos::default())
478            .map_err(|e| {
479                ServerError::Dds(format!(
480                    "rpc client participant for {idl_path} on domain {}: {e:?}",
481                    cfg.domain
482                ))
483            })?;
484        let dispatcher = RpcDispatcher::new(Arc::new(ast), participant)
485            .map_err(|e| ServerError::Config(format!("{idl_path}: {e}")))?;
486        out.push(dispatcher);
487    }
488    Ok(out)
489}
490
491/// Installs a signal watcher for `SIGTERM`/`SIGINT`/`SIGHUP`.
492#[cfg(unix)]
493fn install_signal_watcher(
494    shutdown_flag: Arc<AtomicBool>,
495    reload_flag: Arc<AtomicBool>,
496) -> std::io::Result<JoinHandle<()>> {
497    use signal_hook::consts::{SIGHUP, SIGINT, SIGTERM};
498    use signal_hook::iterator::Signals;
499
500    let mut signals = Signals::new([SIGTERM, SIGINT, SIGHUP])?;
501    thread::Builder::new()
502        .name("zerodds-webdds-signals".into())
503        .spawn(move || {
504            for sig in signals.forever() {
505                match sig {
506                    SIGTERM | SIGINT => {
507                        shutdown_flag.store(true, Ordering::SeqCst);
508                        break;
509                    }
510                    SIGHUP => reload_flag.store(true, Ordering::SeqCst),
511                    _ => {}
512                }
513            }
514        })
515}
516
517#[cfg(not(unix))]
518fn install_signal_watcher(
519    _shutdown_flag: Arc<AtomicBool>,
520    _reload_flag: Arc<AtomicBool>,
521) -> std::io::Result<JoinHandle<()>> {
522    // Windows: signal_hook::iterator is POSIX-only. Spawn a dummy thread,
523    // shutdown runs over the normal socket-close paths.
524    thread::Builder::new()
525        .name("zerodds-webdds-signals".into())
526        .spawn(|| {})
527}
528
529#[cfg(test)]
530#[allow(clippy::unwrap_used, clippy::expect_used)]
531mod tests {
532    use super::*;
533    use std::io::{Read, Write};
534
535    fn free_addr() -> String {
536        let l = TcpListener::bind("127.0.0.1:0").expect("bind");
537        l.local_addr().expect("addr").to_string()
538    }
539
540    #[test]
541    fn daemon_serves_rest_and_wsdl_and_shuts_down() {
542        let mut cfg = DaemonConfig::default_for_dev();
543        cfg.listen = free_addr();
544        let mut handle = start(cfg).expect("start");
545
546        std::thread::sleep(Duration::from_millis(50));
547
548        let mut s = TcpStream::connect(&handle.local_addr).expect("connect");
549        s.set_read_timeout(Some(Duration::from_secs(2))).ok();
550        s.write_all(
551            b"POST /dds/rest1/applications HTTP/1.1\r\nAccept: application/dds-web+xml\r\nContent-Length: 41\r\n\r\n<application><name>app1</name></application>",
552        )
553        .expect("write");
554        let mut buf = [0u8; 4096];
555        let n = s.read(&mut buf).expect("read");
556        let resp = String::from_utf8_lossy(&buf[..n]);
557        assert!(resp.starts_with("HTTP/1.1 201"), "{resp}");
558
559        let mut s2 = TcpStream::connect(&handle.local_addr).expect("connect2");
560        s2.set_read_timeout(Some(Duration::from_secs(2))).ok();
561        s2.write_all(b"GET /wsdl HTTP/1.1\r\nConnection: close\r\n\r\n")
562            .expect("write2");
563        let mut out = Vec::new();
564        s2.read_to_end(&mut out).ok();
565        let resp2 = String::from_utf8_lossy(&out);
566        assert!(resp2.contains("<definitions"), "{resp2}");
567        assert!(resp2.contains("WriteSample"), "{resp2}");
568
569        handle.shutdown();
570        assert!(handle.stop_requested());
571    }
572}
zerodds_webdds_bridge/daemon/server.rs

zerodds_webdds_bridge/daemon/
server.rs
