torrust_server_lib/
registar.rs

1//! Runtime service registry.
2
3use std::collections::HashMap;
4use std::fmt;
5use std::sync::Arc;
6
7use tokio::sync::Mutex;
8use tokio::task::JoinHandle;
9use torrust_net_primitives::service_binding::ServiceBinding;
10
11/// A [`ServiceHeathCheckResult`] is returned by a completed health check.
12pub type ServiceHeathCheckResult = Result<String, String>;
13
14/// The [`ServiceHealthCheckJob`] has a health check job with it's metadata
15///
16/// The `job` awaits a [`ServiceHeathCheckResult`].
17#[derive(Debug)]
18pub struct ServiceHealthCheckJob {
19    pub info: String,
20    pub job: JoinHandle<ServiceHeathCheckResult>,
21}
22
23impl ServiceHealthCheckJob {
24    #[must_use]
25    pub fn new(info: String, job: JoinHandle<ServiceHeathCheckResult>) -> Self {
26        Self { info, job }
27    }
28}
29
30/// The function specification [`FnSpawnServiceHeathCheck`].
31///
32/// A function fulfilling this specification will spawn a new [`ServiceHealthCheckJob`].
33pub type FnSpawnServiceHeathCheck = fn(&ServiceBinding) -> ServiceHealthCheckJob;
34
35/// Immutable data reported by a started local service.
36///
37/// Metadata belongs to the application that uses the registry. The registry
38/// does not assign semantics to it.
39#[derive(Clone, Debug)]
40pub struct ServiceRegistration<M> {
41    service_binding: ServiceBinding,
42    metadata: M,
43    health_check: Option<FnSpawnServiceHeathCheck>,
44}
45
46impl<M> ServiceRegistration<M> {
47    #[must_use]
48    pub fn new(service_binding: ServiceBinding, metadata: M, health_check: Option<FnSpawnServiceHeathCheck>) -> Self {
49        Self {
50            service_binding,
51            metadata,
52            health_check,
53        }
54    }
55
56    #[must_use]
57    pub fn service_binding(&self) -> &ServiceBinding {
58        &self.service_binding
59    }
60
61    #[must_use]
62    pub fn metadata(&self) -> &M {
63        &self.metadata
64    }
65
66    #[must_use]
67    pub fn spawn_check(&self) -> Option<ServiceHealthCheckJob> {
68        self.health_check.map(|health_check| health_check(&self.service_binding))
69    }
70}
71
72/// A cloneable, immutable view of a registered service.
73#[derive(Clone, Debug)]
74pub struct RegisteredService<M> {
75    registration: ServiceRegistration<M>,
76}
77
78impl<M> RegisteredService<M> {
79    #[must_use]
80    pub fn service_binding(&self) -> &ServiceBinding {
81        self.registration.service_binding()
82    }
83
84    #[must_use]
85    pub fn metadata(&self) -> &M {
86        self.registration.metadata()
87    }
88
89    #[must_use]
90    pub fn spawn_check(&self) -> Option<ServiceHealthCheckJob> {
91        self.registration.spawn_check()
92    }
93}
94
95/// Registration failure.
96#[derive(Debug, Clone, PartialEq, Eq)]
97pub enum RegistrationError {
98    /// A service already owns this final local binding.
99    DuplicateBinding(ServiceBinding),
100}
101
102impl fmt::Display for RegistrationError {
103    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
104        match self {
105            Self::DuplicateBinding(service_binding) => {
106                write!(formatter, "a service is already registered for binding {service_binding}")
107            }
108        }
109    }
110}
111
112impl std::error::Error for RegistrationError {}
113
114/// A single-use registration capability for one started service.
115///
116/// Obtain one form per successfully bound service with
117/// [`Registar::give_form`]. Consuming [`Self::register`] acknowledges that the
118/// registration is visible in registry snapshots.
119#[derive(Debug)]
120pub struct ServiceRegistrationForm<M> {
121    registar: Registar<M>,
122}
123
124impl<M> ServiceRegistrationForm<M> {
125    /// Inserts a registration and returns only after it is visible to queries.
126    ///
127    /// A caller may treat successful completion as its registry-readiness
128    /// acknowledgement after binding its listener.
129    ///
130    /// # Errors
131    ///
132    /// Returns [`RegistrationError::DuplicateBinding`] when another service is
133    /// already registered with the same final binding.
134    pub async fn register(self, registration: ServiceRegistration<M>) -> Result<(), RegistrationError> {
135        self.registar.insert(registration).await
136    }
137}
138
139/// The [`Registar`] manages immutable runtime service registrations.
140#[derive(Debug)]
141pub struct Registar<M = ()> {
142    registry: Arc<Mutex<HashMap<ServiceBinding, ServiceRegistration<M>>>>,
143}
144
145impl<M> Clone for Registar<M> {
146    fn clone(&self) -> Self {
147        Self {
148            registry: self.registry.clone(),
149        }
150    }
151}
152
153impl<M> Default for Registar<M> {
154    fn default() -> Self {
155        Self {
156            registry: Arc::default(),
157        }
158    }
159}
160
161impl<M> Registar<M> {
162    /// Returns a capability to register one service.
163    #[must_use]
164    pub fn give_form(&self) -> ServiceRegistrationForm<M> {
165        ServiceRegistrationForm { registar: self.clone() }
166    }
167
168    async fn insert(&self, service_registration: ServiceRegistration<M>) -> Result<(), RegistrationError> {
169        let mut mutex = self.registry.lock().await;
170
171        if mutex.contains_key(service_registration.service_binding()) {
172            return Err(RegistrationError::DuplicateBinding(
173                service_registration.service_binding().clone(),
174            ));
175        }
176
177        mutex.insert(service_registration.service_binding.clone(), service_registration);
178
179        Ok(())
180    }
181
182    /// Returns a deterministic, side-effect-free snapshot of all services.
183    ///
184    /// Results are ordered by protocol, then final socket address, never by
185    /// insertion or hash-map iteration order.
186    pub async fn services(&self) -> Vec<RegisteredService<M>>
187    where
188        M: Clone,
189    {
190        let mutex = self.registry.lock().await;
191        let mut services: Vec<_> = mutex
192            .values()
193            .cloned()
194            .map(|registration| RegisteredService { registration })
195            .collect();
196        services.sort_by(|left, right| {
197            protocol_sort_key(&left.service_binding().protocol())
198                .cmp(&protocol_sort_key(&right.service_binding().protocol()))
199                .then_with(|| {
200                    left.service_binding()
201                        .bind_address()
202                        .cmp(&right.service_binding().bind_address())
203                })
204        });
205        services
206    }
207
208    /// Returns a deterministic, side-effect-free metadata query result.
209    pub async fn services_matching<F>(&self, predicate: F) -> Vec<RegisteredService<M>>
210    where
211        M: Clone,
212        F: Fn(&M) -> bool,
213    {
214        self.services()
215            .await
216            .into_iter()
217            .filter(|service| predicate(service.metadata()))
218            .collect()
219    }
220}
221
222fn protocol_sort_key(protocol: &torrust_net_primitives::service_binding::Protocol) -> u8 {
223    match protocol {
224        torrust_net_primitives::service_binding::Protocol::UDP => 0,
225        torrust_net_primitives::service_binding::Protocol::HTTP => 1,
226        torrust_net_primitives::service_binding::Protocol::HTTPS => 2,
227    }
228}
229
230#[cfg(test)]
231mod tests {
232    use std::net::{Ipv4Addr, SocketAddr};
233
234    use torrust_net_primitives::service_binding::Protocol;
235
236    use super::{Registar, RegistrationError, ServiceRegistration};
237
238    fn binding(protocol: Protocol, port: u16) -> torrust_net_primitives::service_binding::ServiceBinding {
239        torrust_net_primitives::service_binding::ServiceBinding::new(protocol, SocketAddr::from((Ipv4Addr::LOCALHOST, port)))
240            .expect("test binding should be valid")
241    }
242
243    #[tokio::test]
244    async fn it_should_make_a_registration_visible_after_acknowledgement() {
245        let registar = Registar::default();
246
247        registar
248            .give_form()
249            .register(ServiceRegistration::new(binding(Protocol::HTTP, 8000), "first", None))
250            .await
251            .expect("registration should succeed");
252
253        assert_eq!(registar.services().await[0].metadata(), &"first");
254    }
255
256    #[tokio::test]
257    async fn it_should_return_services_in_deterministic_binding_order() {
258        let registar = Registar::default();
259
260        registar
261            .give_form()
262            .register(ServiceRegistration::new(binding(Protocol::HTTP, 9000), "second", None))
263            .await
264            .expect("registration should succeed");
265        registar
266            .give_form()
267            .register(ServiceRegistration::new(binding(Protocol::HTTP, 8000), "first", None))
268            .await
269            .expect("registration should succeed");
270
271        let metadata: Vec<_> = registar
272            .services()
273            .await
274            .into_iter()
275            .map(|service| *service.metadata())
276            .collect();
277
278        assert_eq!(metadata, ["first", "second"]);
279    }
280
281    #[tokio::test]
282    async fn it_should_order_services_by_protocol_then_final_binding() {
283        let registar = Registar::default();
284
285        for (protocol, port, metadata) in [
286            (Protocol::HTTPS, 8000, "https"),
287            (Protocol::HTTP, 9000, "http-second"),
288            (Protocol::UDP, 9000, "udp-second"),
289            (Protocol::HTTP, 8000, "http-first"),
290            (Protocol::UDP, 8000, "udp-first"),
291        ] {
292            registar
293                .give_form()
294                .register(ServiceRegistration::new(binding(protocol, port), metadata, None))
295                .await
296                .expect("registration should succeed");
297        }
298
299        let metadata: Vec<_> = registar
300            .services()
301            .await
302            .into_iter()
303            .map(|service| *service.metadata())
304            .collect();
305
306        assert_eq!(metadata, ["udp-first", "udp-second", "http-first", "http-second", "https"]);
307    }
308
309    #[tokio::test]
310    async fn it_should_reject_duplicate_final_bindings() {
311        let registar = Registar::default();
312        let service_binding = binding(Protocol::HTTP, 8000);
313
314        registar
315            .give_form()
316            .register(ServiceRegistration::new(service_binding.clone(), (), None))
317            .await
318            .expect("initial registration should succeed");
319
320        let error = registar
321            .give_form()
322            .register(ServiceRegistration::new(service_binding.clone(), (), None))
323            .await
324            .expect_err("duplicate registration should fail");
325
326        assert_eq!(error, RegistrationError::DuplicateBinding(service_binding));
327    }
328}
torrust_server_lib/registar.rs

torrust_server_lib/
registar.rs
