cortexfs_channels/
registry.rs

1use std::{collections::BTreeMap, fmt, future::Future, sync::Arc};
2
3use crate::{
4    ChannelAdapter, ChannelError, ChannelFuture, ChannelHealth, ChannelId, ChannelStream,
5    DeliveryReceipt, InboundMessage, OutboundMessage,
6};
7
8/// Thread-safe collection of named channel adapters.
9#[derive(Clone, Default)]
10pub struct ChannelRegistry {
11    adapters: BTreeMap<ChannelId, Arc<dyn ChannelAdapter>>,
12}
13
14impl fmt::Debug for ChannelRegistry {
15    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
16        f.debug_struct("ChannelRegistry")
17            .field("ids", &self.ids())
18            .finish()
19    }
20}
21
22impl ChannelRegistry {
23    #[must_use]
24    pub fn new() -> Self {
25        Self::default()
26    }
27
28    pub fn register(&mut self, adapter: Arc<dyn ChannelAdapter>) -> Result<(), ChannelError> {
29        let id = adapter.id();
30        if self.adapters.contains_key(&id) {
31            return Err(ChannelError::DuplicateChannel(id.to_string()));
32        }
33        self.adapters.insert(id, adapter);
34        Ok(())
35    }
36
37    #[must_use]
38    pub fn ids(&self) -> Vec<ChannelId> {
39        self.adapters.keys().cloned().collect()
40    }
41
42    #[must_use]
43    pub fn get(&self, id: &ChannelId) -> Option<Arc<dyn ChannelAdapter>> {
44        self.adapters.get(id).cloned()
45    }
46
47    pub fn listen(&self, id: &ChannelId) -> Result<ChannelStream, ChannelError> {
48        self.get(id)
49            .ok_or_else(|| ChannelError::UnknownChannel(id.to_string()))?
50            .listen()
51    }
52
53    #[must_use]
54    pub fn send(&self, message: OutboundMessage) -> ChannelFuture<'_, DeliveryReceipt> {
55        Box::pin(async move {
56            let id = message.target.channel.clone();
57            let adapter = self
58                .get(&id)
59                .ok_or_else(|| ChannelError::UnknownChannel(id.to_string()))?;
60            adapter.send(message).await
61        })
62    }
63
64    #[must_use]
65    pub fn health(&self, id: &ChannelId) -> ChannelFuture<'_, ChannelHealth> {
66        let id = id.clone();
67        Box::pin(async move {
68            self.get(&id)
69                .ok_or_else(|| ChannelError::UnknownChannel(id.to_string()))?
70                .health()
71                .await
72        })
73    }
74
75    pub fn dispatch<F, Fut>(
76        &self,
77        inbound: InboundMessage,
78        handler: F,
79    ) -> ChannelFuture<'_, DeliveryReceipt>
80    where
81        F: FnOnce(InboundMessage) -> Fut + Send + 'static,
82        Fut: Future<Output = Result<OutboundMessage, ChannelError>> + Send + 'static,
83    {
84        Box::pin(async move { self.send(handler(inbound).await?).await })
85    }
86}
cortexfs_channels/registry.rs

cortexfs_channels/
registry.rs
