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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
//! The [`ServiceRegistry`]: the daemon's set of hosted services, plus routing.
use std::sync::Arc;
use anyhow::{anyhow, Result};
use serde_json::Value;
use crate::github_rate_limit::{RateLimitCache, RateLimitSnapshot};
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>>,
/// The GitHub API rate-limit monitor's cache (#1375), shared with the poller
/// that fills it (hosted by the worktrees service). `None` until the daemon
/// wires it in via [`set_github_rate_limit`](Self::set_github_rate_limit), so
/// the built-in `status` op can report machine-wide GitHub budget usage as a
/// top-level field without downcasting a specific service.
github_rate_limit: Option<Arc<RateLimitCache>>,
}
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
}
/// Wires in the GitHub rate-limit monitor's cache (#1375), so the built-in
/// `status` op can surface budget usage. Called once at daemon start with the
/// same `Arc` the worktrees service's poller writes.
pub fn set_github_rate_limit(&mut self, cache: Arc<RateLimitCache>) {
self.github_rate_limit = Some(cache);
}
/// The latest GitHub rate-limit snapshot, or `None` when the monitor is
/// unwired or has not polled successfully yet.
#[must_use]
pub fn github_rate_limit(&self) -> Option<RateLimitSnapshot> {
self.github_rate_limit.as_ref().and_then(|c| c.get())
}
/// 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);
}
}