sova_redis/
messaging.rs

1//! Redis Pub/Sub + list queues on [`RedisPool`].
2
3use crate::error::RedisError;
4use crate::pool::RedisPool;
5use futures_util::StreamExt;
6use redis::AsyncCommands;
7use std::time::Duration;
8
9/// One Pub/Sub delivery.
10#[derive(Debug, Clone)]
11pub struct RedisMessage {
12    pub channel: String,
13    pub payload: Vec<u8>,
14    /// Set when received via pattern subscribe (`PSUBSCRIBE`).
15    pub pattern: Option<String>,
16}
17
18impl RedisMessage {
19    pub fn payload_str(&self) -> Option<&str> {
20        std::str::from_utf8(&self.payload).ok()
21    }
22}
23
24/// Dedicated Pub/Sub connection (not shared with normal commands).
25pub struct RedisSubscriber {
26    pubsub: redis::aio::PubSub,
27}
28
29impl RedisSubscriber {
30    /// Wait for the next message (`None` when the connection closes).
31    pub async fn next(&mut self) -> Option<RedisMessage> {
32        let msg = self.pubsub.on_message().next().await?;
33        let pattern = msg
34            .get_pattern::<Option<String>>()
35            .ok()
36            .flatten()
37            .filter(|s| !s.is_empty());
38        Some(RedisMessage {
39            channel: msg.get_channel_name().to_string(),
40            payload: msg.get_payload_bytes().to_vec(),
41            pattern,
42        })
43    }
44
45    pub async fn subscribe(&mut self, channel: impl AsRef<str>) -> Result<(), RedisError> {
46        self.pubsub
47            .subscribe(channel.as_ref())
48            .await
49            .map_err(RedisError::from)
50    }
51
52    pub async fn unsubscribe(&mut self, channel: impl AsRef<str>) -> Result<(), RedisError> {
53        self.pubsub
54            .unsubscribe(channel.as_ref())
55            .await
56            .map_err(RedisError::from)
57    }
58
59    pub async fn psubscribe(&mut self, pattern: impl AsRef<str>) -> Result<(), RedisError> {
60        self.pubsub
61            .psubscribe(pattern.as_ref())
62            .await
63            .map_err(RedisError::from)
64    }
65
66    pub async fn punsubscribe(&mut self, pattern: impl AsRef<str>) -> Result<(), RedisError> {
67        self.pubsub
68            .punsubscribe(pattern.as_ref())
69            .await
70            .map_err(RedisError::from)
71    }
72}
73
74fn rid() -> Option<String> {
75    sova_core::current_request_id()
76}
77
78fn trunc(s: &str) -> String {
79    const MAX: usize = 120;
80    if s.len() <= MAX {
81        s.to_string()
82    } else {
83        format!("{}…", &s[..MAX - 1])
84    }
85}
86
87impl RedisPool {
88    /// `PUBLISH channel message` — returns subscriber count that received it.
89    pub async fn publish(
90        &self,
91        channel: impl AsRef<str>,
92        message: impl AsRef<[u8]>,
93    ) -> Result<i64, RedisError> {
94        let started = std::time::Instant::now();
95        let ch = channel.as_ref();
96        let res = if let Some(fake) = self.fake() {
97            fake.publish(ch, message.as_ref())
98        } else {
99            let mut conn = self.get()?;
100            conn.publish(ch, message.as_ref())
101                .await
102                .map_err(RedisError::from)
103        };
104        let ok = res.is_ok();
105        tracing::debug!(
106            target: "sova.redis",
107            cmd = "publish",
108            key = %trunc(ch),
109            channel = %trunc(ch),
110            ok,
111            duration_ms = started.elapsed().as_secs_f64() * 1000.0,
112            request_id = rid().as_deref().unwrap_or(""),
113            "sova.redis"
114        );
115        res
116    }
117
118    /// Open a dedicated Pub/Sub connection and subscribe to `channels`.
119    ///
120    /// Requires the pool URL (set by the [`crate::Redis`] plugin). Do not reuse
121    /// the shared command connection for subscribe.
122    pub async fn subscribe(
123        &self,
124        channels: impl IntoIterator<Item = impl AsRef<str>>,
125    ) -> Result<RedisSubscriber, RedisError> {
126        if self.is_fake() {
127            return Err(RedisError::msg(
128                "fake redis: use subscribe_fake() for a single channel",
129            ));
130        }
131        let url = self.url()?;
132        let client = redis::Client::open(url.as_str()).map_err(RedisError::from)?;
133        let mut pubsub = client.get_async_pubsub().await.map_err(RedisError::from)?;
134        let names: Vec<String> = channels
135            .into_iter()
136            .map(|c| c.as_ref().to_string())
137            .collect();
138        if names.is_empty() {
139            return Err(RedisError::msg("subscribe: no channels"));
140        }
141        pubsub.subscribe(&names).await.map_err(RedisError::from)?;
142        Ok(RedisSubscriber { pubsub })
143    }
144
145    /// Pattern subscribe (`PSUBSCRIBE`), e.g. `events:*`.
146    pub async fn psubscribe(
147        &self,
148        patterns: impl IntoIterator<Item = impl AsRef<str>>,
149    ) -> Result<RedisSubscriber, RedisError> {
150        let url = self.url()?;
151        let client = redis::Client::open(url.as_str()).map_err(RedisError::from)?;
152        let mut pubsub = client.get_async_pubsub().await.map_err(RedisError::from)?;
153        let names: Vec<String> = patterns
154            .into_iter()
155            .map(|c| c.as_ref().to_string())
156            .collect();
157        if names.is_empty() {
158            return Err(RedisError::msg("psubscribe: no patterns"));
159        }
160        pubsub.psubscribe(&names).await.map_err(RedisError::from)?;
161        Ok(RedisSubscriber { pubsub })
162    }
163
164    /// Push onto a list queue (`LPUSH`). Returns new list length.
165    pub async fn enqueue(
166        &self,
167        queue: impl AsRef<str>,
168        message: impl AsRef<[u8]>,
169    ) -> Result<i64, RedisError> {
170        let started = std::time::Instant::now();
171        let q = queue.as_ref();
172        let mut conn = self.get()?;
173        let res = conn
174            .lpush(q, message.as_ref())
175            .await
176            .map_err(RedisError::from);
177        let ok = res.is_ok();
178        tracing::debug!(
179            target: "sova.redis",
180            cmd = "enqueue",
181            key = %trunc(q),
182            queue = %trunc(q),
183            ok,
184            duration_ms = started.elapsed().as_secs_f64() * 1000.0,
185            request_id = rid().as_deref().unwrap_or(""),
186            "sova.redis"
187        );
188        res
189    }
190
191    /// Non-blocking pop (`RPOP`). `None` if empty.
192    pub async fn dequeue(&self, queue: impl AsRef<str>) -> Result<Option<Vec<u8>>, RedisError> {
193        let started = std::time::Instant::now();
194        let q = queue.as_ref();
195        let mut conn = self.get()?;
196        let res: Result<Option<Vec<u8>>, RedisError> =
197            conn.rpop(q, None).await.map_err(RedisError::from);
198        let ok = res.is_ok();
199        let hit = matches!(res, Ok(Some(_)));
200        tracing::debug!(
201            target: "sova.redis",
202            cmd = "dequeue",
203            key = %trunc(q),
204            queue = %trunc(q),
205            hit,
206            ok,
207            duration_ms = started.elapsed().as_secs_f64() * 1000.0,
208            request_id = rid().as_deref().unwrap_or(""),
209            "sova.redis"
210        );
211        res
212    }
213
214    /// Blocking pop (`BRPOP`). `timeout = 0` waits forever.
215    ///
216    /// Returns `(queue_name, payload)` or `None` on timeout.
217    pub async fn dequeue_wait(
218        &self,
219        queue: impl AsRef<str>,
220        timeout: Duration,
221    ) -> Result<Option<(String, Vec<u8>)>, RedisError> {
222        let started = std::time::Instant::now();
223        let q = queue.as_ref();
224        let mut conn = self.get()?;
225        let secs = timeout.as_secs() as f64 + f64::from(timeout.subsec_nanos()) / 1e9;
226        let res: Result<Option<(String, Vec<u8>)>, RedisError> = redis::cmd("BRPOP")
227            .arg(q)
228            .arg(secs)
229            .query_async(&mut conn)
230            .await
231            .map_err(RedisError::from);
232        let ok = res.is_ok();
233        let hit = matches!(res, Ok(Some(_)));
234        tracing::debug!(
235            target: "sova.redis",
236            cmd = "dequeue_wait",
237            key = %trunc(q),
238            queue = %trunc(q),
239            hit,
240            ok,
241            duration_ms = started.elapsed().as_secs_f64() * 1000.0,
242            request_id = rid().as_deref().unwrap_or(""),
243            "sova.redis"
244        );
245        res
246    }
247}
sova_redis/messaging.rs

sova_redis/
messaging.rs
