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