aster_forge_events/
redis_transport.rs

1use async_trait::async_trait;
2use futures::StreamExt;
3use redis::AsyncCommands;
4use std::future::Future;
5use std::sync::Arc;
6use tokio::sync::mpsc;
7use tokio_util::sync::CancellationToken;
8
9use crate::{
10    EventConnectionObservation, EventReconnectPolicy, EventSubscriptionSource,
11    EventSubscriptionUpdate, supervise_event_subscription,
12};
13
14/// Redis event transport errors returned while creating a publisher.
15#[derive(Debug, thiserror::Error)]
16pub enum RedisEventBusError {
17    /// The Redis client could not be created from the configured URL.
18    #[error("open Redis event URL: {0}")]
19    Open(String),
20    /// A configured event topic is empty.
21    #[error("event topic must not be empty")]
22    EmptyTopic,
23    /// Publishing failed.
24    #[error("publish event payload: {0}")]
25    Publish(String),
26    /// Opening a subscription failed.
27    #[error("subscribe to event topic: {0}")]
28    Subscribe(String),
29    /// An active subscription ended unexpectedly.
30    #[error("event subscription stream ended")]
31    StreamEnded,
32}
33
34/// Receives subscriber lifecycle observations.
35pub trait EventConnectionObserver: Send + Sync {
36    /// Records one connection transition.
37    fn observe_event_connection(&self, observation: EventConnectionObservation);
38}
39
40impl<F> EventConnectionObserver for F
41where
42    F: Fn(EventConnectionObservation) + Send + Sync,
43{
44    fn observe_event_connection(&self, observation: EventConnectionObservation) {
45        self(observation);
46    }
47}
48
49/// Backwards-compatible name for the shared event reconnect policy.
50pub type RedisEventReconnectPolicy = EventReconnectPolicy;
51
52/// Redis-backed transient event publisher and reconnecting subscriber.
53#[derive(Clone)]
54pub struct RedisEventBus {
55    client: redis::Client,
56    topic: String,
57    reconnect_policy: EventReconnectPolicy,
58}
59
60/// One active Redis Pub/Sub subscription.
61pub struct RedisEventSubscription {
62    pubsub: redis::aio::PubSub,
63}
64
65impl RedisEventBus {
66    /// Creates a bus from a Redis URL and logical topic.
67    pub fn from_url(url: &str, topic: impl Into<String>) -> Result<Self, RedisEventBusError> {
68        let topic = topic.into();
69        if topic.trim().is_empty() {
70            return Err(RedisEventBusError::EmptyTopic);
71        }
72        let client = redis::Client::open(url)
73            .map_err(|error| RedisEventBusError::Open(error.to_string()))?;
74        Ok(Self {
75            client,
76            topic,
77            reconnect_policy: EventReconnectPolicy::default(),
78        })
79    }
80
81    /// Creates a bus from an existing Redis client.
82    pub fn from_client(client: redis::Client, topic: impl Into<String>) -> Self {
83        Self {
84            client,
85            topic: topic.into(),
86            reconnect_policy: EventReconnectPolicy::default(),
87        }
88    }
89
90    /// Overrides the reconnect policy.
91    pub fn with_reconnect_policy(mut self, policy: RedisEventReconnectPolicy) -> Self {
92        self.reconnect_policy = policy;
93        self
94    }
95
96    /// Returns the configured logical topic.
97    pub fn topic(&self) -> &str {
98        &self.topic
99    }
100
101    /// Publishes one opaque payload. Payload interpretation belongs to the product layer.
102    pub async fn publish(&self, payload: impl Into<String>) -> Result<(), RedisEventBusError> {
103        let mut connection = self
104            .client
105            .get_multiplexed_async_connection()
106            .await
107            .map_err(|error| RedisEventBusError::Publish(error.to_string()))?;
108        let _: usize = connection
109            .publish(&self.topic, payload.into())
110            .await
111            .map_err(|error| RedisEventBusError::Publish(error.to_string()))?;
112        Ok(())
113    }
114
115    /// Opens one Redis subscription attempt.
116    pub async fn subscribe(&self) -> Result<RedisEventSubscription, RedisEventBusError> {
117        let mut pubsub = self
118            .client
119            .get_async_pubsub()
120            .await
121            .map_err(|error| RedisEventBusError::Subscribe(error.to_string()))?;
122        pubsub
123            .subscribe(&self.topic)
124            .await
125            .map_err(|error| RedisEventBusError::Subscribe(error.to_string()))?;
126        Ok(RedisEventSubscription { pubsub })
127    }
128
129    /// Runs a reconnecting subscription until shutdown is cancelled.
130    ///
131    /// Malformed Redis payloads are logged and skipped. The callback is responsible for decoding
132    /// the product payload and deciding whether an event belongs to the current runtime.
133    pub async fn run_subscription<F, Fut>(
134        &self,
135        shutdown: CancellationToken,
136        observer: Option<&dyn EventConnectionObserver>,
137        mut on_payload: F,
138    ) where
139        F: FnMut(String) -> Fut,
140        Fut: Future<Output = ()>,
141    {
142        let (updates_tx, mut updates_rx) = mpsc::channel(1);
143        let supervisor = supervise_event_subscription(
144            Arc::new(self.clone()),
145            self.reconnect_policy,
146            shutdown.clone(),
147            updates_tx,
148        );
149        tokio::pin!(supervisor);
150
151        loop {
152            let update = tokio::select! {
153                () = shutdown.cancelled() => return,
154                () = &mut supervisor => return,
155                update = updates_rx.recv() => update,
156            };
157            match update {
158                Some(EventSubscriptionUpdate::Connection(observation)) => {
159                    if let Some(observer) = observer {
160                        observer.observe_event_connection(observation);
161                    }
162                }
163                Some(EventSubscriptionUpdate::Item(payload)) => on_payload(payload).await,
164                None => return,
165            }
166        }
167    }
168}
169
170impl RedisEventSubscription {
171    /// Receives one raw payload. Redis payload conversion failures are logged and skipped.
172    pub async fn receive(&mut self) -> Result<String, RedisEventBusError> {
173        loop {
174            let mut stream = self.pubsub.on_message();
175            let Some(message) = stream.next().await else {
176                return Err(RedisEventBusError::StreamEnded);
177            };
178            match message.get_payload::<String>() {
179                Ok(payload) => return Ok(payload),
180                Err(error) => tracing::warn!(%error, "failed to decode Redis event payload"),
181            }
182        }
183    }
184}
185
186#[async_trait]
187impl EventSubscriptionSource for RedisEventBus {
188    type Item = String;
189    type Subscription = RedisEventSubscription;
190    type Error = RedisEventBusError;
191
192    async fn subscribe(&self) -> Result<Self::Subscription, Self::Error> {
193        RedisEventBus::subscribe(self).await
194    }
195
196    async fn receive(
197        &self,
198        subscription: &mut Self::Subscription,
199    ) -> Result<Self::Item, Self::Error> {
200        subscription.receive().await
201    }
202}
203
204#[cfg(test)]
205mod tests {
206    use super::{RedisEventBus, RedisEventBusError};
207
208    #[test]
209    fn rejects_empty_topics() {
210        assert!(matches!(
211            RedisEventBus::from_url("redis://127.0.0.1", "  "),
212            Err(RedisEventBusError::EmptyTopic)
213        ));
214    }
215}
aster_forge_events/redis_transport.rs

aster_forge_events/
redis_transport.rs
