aster_forge_config/notification/
notifier.rs

1use async_trait::async_trait;
2use std::sync::Arc;
3use tokio::sync::broadcast;
4
5use crate::{ConfigCoreError, Result};
6
7#[cfg(feature = "redis-pubsub")]
8use super::message::decode_config_reload_transport_payload;
9use super::message::{ConfigChangeEvent, ConfigReloadMessage};
10
11/// Subscription returned by config notifiers.
12pub struct ConfigNotification {
13    receiver: ConfigNotificationReceiver,
14}
15
16enum ConfigNotificationReceiver {
17    InMemory(broadcast::Receiver<ConfigChangeEvent>),
18    #[cfg(feature = "redis-pubsub")]
19    Redis(aster_forge_events::RedisEventSubscription),
20}
21
22impl ConfigNotification {
23    pub(super) fn new(receiver: broadcast::Receiver<ConfigChangeEvent>) -> Self {
24        Self {
25            receiver: ConfigNotificationReceiver::InMemory(receiver),
26        }
27    }
28
29    #[cfg(feature = "redis-pubsub")]
30    fn from_redis(subscription: aster_forge_events::RedisEventSubscription) -> Self {
31        Self {
32            receiver: ConfigNotificationReceiver::Redis(subscription),
33        }
34    }
35
36    /// Waits for the next notification.
37    pub async fn recv(&mut self) -> Result<ConfigChangeEvent> {
38        match &mut self.receiver {
39            ConfigNotificationReceiver::InMemory(receiver) => receiver
40                .recv()
41                .await
42                .map_err(|error| ConfigCoreError::notification(error.to_string())),
43            #[cfg(feature = "redis-pubsub")]
44            ConfigNotificationReceiver::Redis(subscription) => loop {
45                let payload = subscription
46                    .receive()
47                    .await
48                    .map_err(|error| ConfigCoreError::notification(error.to_string()))?;
49                match decode_config_reload_transport_payload(&payload) {
50                    Ok(event) => return Ok(event),
51                    Err(error) => {
52                        tracing::warn!(%error, "failed to parse Redis config reload message");
53                    }
54                }
55            },
56        }
57    }
58}
59
60/// Transport used to publish and subscribe to reload notifications.
61#[async_trait]
62pub trait ConfigChangeNotifier: Send + Sync {
63    /// Publishes a reload notification.
64    async fn publish_reload(&self, message: ConfigReloadMessage) -> Result<()>;
65
66    /// Subscribes to future reload notifications.
67    async fn subscribe(&self) -> Result<ConfigNotification>;
68}
69
70/// Shared notifier object used by runtime services.
71pub type SharedConfigChangeNotifier = Arc<dyn ConfigChangeNotifier>;
72
73/// In-memory notifier for single-process deployments and tests.
74#[derive(Debug, Clone)]
75pub struct InMemoryConfigNotifier {
76    pub(super) sender: broadcast::Sender<ConfigChangeEvent>,
77}
78
79impl InMemoryConfigNotifier {
80    /// Creates a notifier with the given broadcast channel capacity.
81    pub fn new(capacity: usize) -> Self {
82        let (sender, _) = broadcast::channel(capacity.max(1));
83        Self { sender }
84    }
85}
86
87impl Default for InMemoryConfigNotifier {
88    fn default() -> Self {
89        Self::new(128)
90    }
91}
92
93#[async_trait]
94impl ConfigChangeNotifier for InMemoryConfigNotifier {
95    async fn publish_reload(&self, message: ConfigReloadMessage) -> Result<()> {
96        // broadcast::Sender::send fails only when no receivers exist. A reload
97        // notification nobody is listening to is not an error: the change itself
98        // already succeeded, and single-process deployments may legitimately run
99        // without a subscription worker.
100        let _ = self.sender.send(ConfigChangeEvent::Reload(message));
101        Ok(())
102    }
103
104    async fn subscribe(&self) -> Result<ConfigNotification> {
105        Ok(ConfigNotification::new(self.sender.subscribe()))
106    }
107}
108
109#[cfg(feature = "redis-pubsub")]
110mod redis_transport {
111    use super::{ConfigChangeNotifier, ConfigNotification, ConfigReloadMessage};
112    use crate::{ConfigCoreError, Result};
113
114    /// Redis pub/sub publisher for configuration reload messages.
115    #[derive(Clone)]
116    pub struct RedisConfigChangeNotifier {
117        bus: aster_forge_events::RedisEventBus,
118    }
119
120    impl RedisConfigChangeNotifier {
121        /// Creates a Redis notifier for `channel`.
122        pub fn new(client: redis::Client, channel: impl Into<String>) -> Self {
123            Self {
124                bus: aster_forge_events::RedisEventBus::from_client(client, channel),
125            }
126        }
127
128        /// Creates a Redis notifier from a Redis connection URL.
129        pub fn from_url(url: &str, channel: impl Into<String>) -> Result<Self> {
130            let bus = aster_forge_events::RedisEventBus::from_url(url, channel)
131                .map_err(|error| ConfigCoreError::notification(error.to_string()))?;
132            Ok(Self { bus })
133        }
134    }
135
136    #[async_trait::async_trait]
137    impl ConfigChangeNotifier for RedisConfigChangeNotifier {
138        async fn publish_reload(&self, message: ConfigReloadMessage) -> Result<()> {
139            let payload = message.encode()?;
140            self.bus
141                .publish(payload)
142                .await
143                .map_err(|error| ConfigCoreError::notification(error.to_string()))
144        }
145
146        async fn subscribe(&self) -> Result<ConfigNotification> {
147            let subscription = self
148                .bus
149                .subscribe()
150                .await
151                .map_err(|error| ConfigCoreError::notification(error.to_string()))?;
152            Ok(ConfigNotification::from_redis(subscription))
153        }
154    }
155}
156
157#[cfg(feature = "redis-pubsub")]
158pub use redis_transport::RedisConfigChangeNotifier;
aster_forge_config/notification/notifier.rs

aster_forge_config/notification/
notifier.rs
