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
//! LIM-002 R6: the channel registry.
//!
//! [`ChannelRegistry`] maps channel names to their supervised actor handles on a
//! single shared [`ChannelSupervisor`]. It supports `create` (spawn a new
//! channel actor), `lookup` (by name), `list` (all active channels with their
//! current subscriber counts), and `close` (signal an actor to shut down and
//! drop its registry entry). Duplicate names are rejected.
use std::collections::HashMap;
use std::sync::Mutex;
use crate::channel::supervisor::ChannelSupervisor;
use crate::channel::types::{ChannelConfig, ChannelHandle};
use crate::error::LiminalError;
/// One active channel's name and current subscriber count.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ChannelSummary {
/// The channel's registered name.
pub name: String,
/// Number of currently-linked subscribers on the channel actor.
pub subscriber_count: usize,
}
/// A name-keyed registry of supervised channel actors.
#[derive(Debug)]
pub struct ChannelRegistry {
supervisor: ChannelSupervisor,
channels: Mutex<HashMap<String, ChannelHandle>>,
}
impl ChannelRegistry {
/// Builds a registry backed by a dedicated supervisor (its own scheduler).
///
/// # Errors
/// Returns [`LiminalError`] when the supervisor's scheduler cannot start.
pub fn new() -> Result<Self, LiminalError> {
Ok(Self {
supervisor: ChannelSupervisor::new()?,
channels: Mutex::new(HashMap::new()),
})
}
/// Builds a registry over an existing supervisor.
#[must_use]
pub fn with_supervisor(supervisor: ChannelSupervisor) -> Self {
Self {
supervisor,
channels: Mutex::new(HashMap::new()),
}
}
/// Creates and registers a new channel actor for `config`, returning its
/// handle. Rejects a name already present in the registry.
///
/// # Errors
/// Returns [`LiminalError::PublishFailed`] when `config.name` is already
/// registered, or a [`LiminalError`] when the registry lock is poisoned.
pub fn create(&self, config: ChannelConfig) -> Result<ChannelHandle, LiminalError> {
let mut channels = self.lock()?;
if channels.contains_key(&config.name) {
return Err(LiminalError::PublishFailed {
message: format!("channel '{}' already exists", config.name),
});
}
let name = config.name.clone();
let handle = ChannelHandle::with_supervisor(config, self.supervisor.clone());
channels.insert(name, handle.clone());
drop(channels);
Ok(handle)
}
/// Looks up a registered channel handle by name.
///
/// # Errors
/// Returns [`LiminalError`] when the registry lock is poisoned.
pub fn lookup(&self, name: &str) -> Result<Option<ChannelHandle>, LiminalError> {
Ok(self.lock()?.get(name).cloned())
}
/// Lists every active channel with its current subscriber count.
///
/// The subscriber count is read live from each actor (`ListSubscribers`); a
/// channel whose actor is unreachable reports a zero count rather than
/// failing the whole listing.
///
/// # Errors
/// Returns [`LiminalError`] when the registry lock is poisoned.
pub fn list(&self) -> Result<Vec<ChannelSummary>, LiminalError> {
let snapshot: Vec<(String, ChannelHandle)> = {
let channels = self.lock()?;
channels
.iter()
.map(|(name, handle)| (name.clone(), handle.clone()))
.collect()
};
let mut summaries = snapshot
.into_iter()
.map(|(name, handle)| ChannelSummary {
subscriber_count: handle.subscriber_count().unwrap_or(0),
name,
})
.collect::<Vec<_>>();
summaries.sort_by(|left, right| left.name.cmp(&right.name));
Ok(summaries)
}
/// Closes the named channel (stopping its actor process) and drops its
/// registry entry. Returns whether a channel was present.
///
/// # Errors
/// Returns [`LiminalError`] when the registry lock is poisoned or the close
/// command fails.
pub fn close(&self, name: &str) -> Result<bool, LiminalError> {
let handle = self.lock()?.remove(name);
match handle {
Some(handle) => {
handle.close()?;
Ok(true)
}
None => Ok(false),
}
}
/// Stops the underlying supervisor scheduler.
pub fn shutdown(&self) {
self.supervisor.shutdown();
}
fn lock(
&self,
) -> Result<std::sync::MutexGuard<'_, HashMap<String, ChannelHandle>>, LiminalError> {
self.channels
.lock()
.map_err(|error| LiminalError::PublishFailed {
message: format!("channel registry lock poisoned: {error}"),
})
}
}