omni_dev/daemon/
registry.rs

1//! The [`ServiceRegistry`]: the daemon's set of hosted services, plus routing.
2
3use std::sync::Arc;
4
5use anyhow::{anyhow, Result};
6use serde_json::Value;
7
8use super::service::{DaemonService, ServiceStatus, ServiceStream};
9
10/// Holds the daemon's registered services and routes control-socket envelopes
11/// to them by [`name`](DaemonService::name).
12#[derive(Clone, Default)]
13pub struct ServiceRegistry {
14    services: Vec<Arc<dyn DaemonService>>,
15}
16
17impl ServiceRegistry {
18    /// Creates an empty registry.
19    pub fn new() -> Self {
20        Self::default()
21    }
22
23    /// Adds a service. Lookups match by [`DaemonService::name`] and the first
24    /// registration wins; a second service sharing a name would be dead code
25    /// for routing and would double-count in status/menu iteration, so it is
26    /// rejected with a warning rather than silently kept.
27    pub fn register(&mut self, service: Arc<dyn DaemonService>) {
28        let name = service.name();
29        if self.services.iter().any(|s| s.name() == name) {
30            tracing::warn!("ignoring duplicate registration of daemon service `{name}`");
31            return;
32        }
33        self.services.push(service);
34    }
35
36    /// Returns the registered service with the given name, if any.
37    pub fn get(&self, name: &str) -> Option<&Arc<dyn DaemonService>> {
38        self.services.iter().find(|s| s.name() == name)
39    }
40
41    /// All registered services, in registration order.
42    pub fn services(&self) -> &[Arc<dyn DaemonService>] {
43        &self.services
44    }
45
46    /// Routes an operation to the named service, erroring if no such service is
47    /// registered.
48    pub async fn dispatch(&self, service: &str, op: &str, payload: Value) -> Result<Value> {
49        let svc = self
50            .get(service)
51            .ok_or_else(|| anyhow!("unknown service: {service}"))?;
52        svc.handle(op, payload).await
53    }
54
55    /// Opens a push subscription on the named service for a streaming `op`, or
56    /// `None` when the service is unknown or does not stream that op — in which
57    /// case the caller falls back to the normal [`dispatch`](Self::dispatch)
58    /// request→reply path (#1267).
59    pub fn subscribe(
60        &self,
61        service: &str,
62        op: &str,
63        payload: &Value,
64    ) -> Option<Box<dyn ServiceStream>> {
65        self.get(service)?.subscribe(op, payload)
66    }
67
68    /// Collects status from every service, in registration order.
69    pub async fn statuses(&self) -> Vec<ServiceStatus> {
70        let mut out = Vec::with_capacity(self.services.len());
71        for svc in &self.services {
72            out.push(svc.status().await);
73        }
74        out
75    }
76
77    /// Gracefully shuts down every service, in registration order.
78    pub async fn shutdown_all(&self) {
79        for svc in &self.services {
80            svc.shutdown().await;
81        }
82    }
83}
84
85#[cfg(test)]
86#[allow(clippy::unwrap_used, clippy::expect_used)]
87mod tests {
88    use super::*;
89    use crate::daemon::services::echo::EchoService;
90    use serde_json::json;
91
92    #[tokio::test]
93    async fn routes_known_service_and_rejects_unknown() {
94        let mut registry = ServiceRegistry::new();
95        assert!(registry.services().is_empty());
96        registry.register(Arc::new(EchoService));
97
98        assert!(registry.get("echo").is_some());
99        assert!(registry.get("missing").is_none());
100
101        // Routed op reaches the service; an unknown service is an error.
102        assert_eq!(
103            registry
104                .dispatch("echo", "echo", json!({ "x": 1 }))
105                .await
106                .unwrap(),
107            json!({ "x": 1 })
108        );
109        let err = registry
110            .dispatch("missing", "echo", Value::Null)
111            .await
112            .unwrap_err();
113        assert!(err.to_string().contains("unknown service"));
114
115        // Aggregation iterates every registered service.
116        assert_eq!(registry.statuses().await.len(), 1);
117        registry.shutdown_all().await;
118    }
119
120    #[test]
121    fn duplicate_registration_is_ignored() {
122        let mut registry = ServiceRegistry::new();
123        registry.register(Arc::new(EchoService));
124        registry.register(Arc::new(EchoService));
125        // The second registration shares `echo`'s name, so it is dropped: the
126        // first wins on lookup and iteration is not double-counted.
127        assert_eq!(registry.services().len(), 1);
128    }
129}
omni_dev/daemon/registry.rs

omni_dev/daemon/
registry.rs
