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    ///
38    /// # Errors
39    ///
40    /// Returns [`ConfigError`] when the notifier subscription cannot receive or decode the next event.
41    pub async fn recv(&mut self) -> Result<ConfigChangeEvent> {
42        match &mut self.receiver {
43            ConfigNotificationReceiver::InMemory(receiver) => receiver
44                .recv()
45                .await
46                .map_err(|error| ConfigCoreError::notification(error.to_string())),
47            #[cfg(feature = "redis-pubsub")]
48            ConfigNotificationReceiver::Redis(subscription) => loop {
49                let payload = subscription
50                    .receive()
51                    .await
52                    .map_err(|error| ConfigCoreError::notification(error.to_string()))?;
53                match decode_config_reload_transport_payload(&payload) {
54                    Ok(event) => return Ok(event),
55                    Err(error) => {
56                        tracing::warn!(%error, "failed to parse Redis config reload message");
57                    }
58                }
59            },
60        }
61    }
62}
63
64/// Transport used to publish and subscribe to reload notifications.
65#[async_trait]
66pub trait ConfigChangeNotifier: Send + Sync {
67    /// Publishes a reload notification.
68    async fn publish_reload(&self, message: ConfigReloadMessage) -> Result<()>;
69
70    /// Subscribes to future reload notifications.
71    async fn subscribe(&self) -> Result<ConfigNotification>;
72}
73
74/// Shared notifier object used by runtime services.
75pub type SharedConfigChangeNotifier = Arc<dyn ConfigChangeNotifier>;
76
77/// In-memory notifier for single-process deployments and tests.
78#[derive(Debug, Clone)]
79pub struct InMemoryConfigNotifier {
80    pub(super) sender: broadcast::Sender<ConfigChangeEvent>,
81}
82
83impl InMemoryConfigNotifier {
84    /// Creates a notifier with the given broadcast channel capacity.
85    #[must_use]
86    pub fn new(capacity: usize) -> Self {
87        let (sender, _) = broadcast::channel(capacity.max(1));
88        Self { sender }
89    }
90}
91
92impl Default for InMemoryConfigNotifier {
93    fn default() -> Self {
94        Self::new(128)
95    }
96}
97
98#[async_trait]
99impl ConfigChangeNotifier for InMemoryConfigNotifier {
100    async fn publish_reload(&self, message: ConfigReloadMessage) -> Result<()> {
101        // broadcast::Sender::send fails only when no receivers exist. A reload
102        // notification nobody is listening to is not an error: the change itself
103        // already succeeded, and single-process deployments may legitimately run
104        // without a subscription worker.
105        let _ = self.sender.send(ConfigChangeEvent::Reload(message));
106        Ok(())
107    }
108
109    async fn subscribe(&self) -> Result<ConfigNotification> {
110        Ok(ConfigNotification::new(self.sender.subscribe()))
111    }
112}
113
114#[cfg(feature = "redis-pubsub")]
115mod redis_transport {
116    use super::{ConfigChangeNotifier, ConfigNotification, ConfigReloadMessage};
117    use crate::{ConfigCoreError, Result};
118
119    /// Redis pub/sub publisher for configuration reload messages.
120    #[derive(Clone)]
121    pub struct RedisConfigChangeNotifier {
122        bus: aster_forge_events::RedisEventBus,
123    }
124
125    impl RedisConfigChangeNotifier {
126        /// Creates a Redis notifier for `channel`.
127        pub fn new(client: redis::Client, channel: impl Into<String>) -> Self {
128            Self {
129                bus: aster_forge_events::RedisEventBus::from_client(client, channel),
130            }
131        }
132
133        /// Creates a Redis notifier from a Redis connection URL.
134        ///
135        /// # Errors
136        ///
137        /// Returns [`ConfigError`] when the Redis endpoint, credentials, or notifier connection is invalid.
138        pub fn from_url(url: &str, channel: impl Into<String>) -> Result<Self> {
139            let bus = aster_forge_events::RedisEventBus::from_url(url, channel)
140                .map_err(|error| ConfigCoreError::notification(error.to_string()))?;
141            Ok(Self { bus })
142        }
143
144        /// Creates a Redis notifier from a base URL and raw credentials.
145        ///
146        /// # Errors
147        ///
148        /// Returns [`ConfigError`] when the Redis endpoint, credentials, or notifier connection is invalid.
149        pub fn from_credentials(
150            base_url: &str,
151            username: Option<&str>,
152            password: Option<&str>,
153            channel: impl Into<String>,
154        ) -> Result<Self> {
155            let bus = aster_forge_events::RedisEventBus::from_credentials(
156                base_url, username, password, channel,
157            )
158            .map_err(|error| ConfigCoreError::notification(error.to_string()))?;
159            Ok(Self { bus })
160        }
161    }
162
163    #[async_trait::async_trait]
164    impl ConfigChangeNotifier for RedisConfigChangeNotifier {
165        async fn publish_reload(&self, message: ConfigReloadMessage) -> Result<()> {
166            let payload = message.encode()?;
167            self.bus
168                .publish(payload)
169                .await
170                .map_err(|error| ConfigCoreError::notification(error.to_string()))
171        }
172
173        async fn subscribe(&self) -> Result<ConfigNotification> {
174            let subscription = self
175                .bus
176                .subscribe()
177                .await
178                .map_err(|error| ConfigCoreError::notification(error.to_string()))?;
179            Ok(ConfigNotification::from_redis(subscription))
180        }
181    }
182}
183
184#[cfg(feature = "redis-pubsub")]
185pub use redis_transport::RedisConfigChangeNotifier;
aster_forge_config/notification/notifier.rs

aster_forge_config/notification/
notifier.rs
