kevy_resp/pubsub_event.rs
1//! The pubsub frame vocabulary shared by every kevy client crate and the
2//! embedded engine's in-process bus: [`PubsubEvent`] (one received frame,
3//! acks and deliveries alike)
4//! and its `TryFrom<Reply>` (RESP reply → event, handling both RESP2
5//! `*N` arrays and RESP3 `>N` push frames).
6
7use std::io;
8
9use crate::Reply;
10
11/// One pubsub frame received from the bus or the wire.
12///
13/// `Unsubscribe` / `Punsubscribe`'s `channel` / `pattern` is `None`
14/// when the server acknowledges "unsubscribed from everything" with a
15/// nil bulk — matching the Redis wire shape.
16///
17/// ```
18/// use kevy_resp::{PubsubEvent, parse_reply};
19///
20/// let (reply, _) = parse_reply(b"*3\r\n$7\r\nmessage\r\n$4\r\nnews\r\n$2\r\nhi\r\n")?
21/// .expect("complete frame");
22/// match PubsubEvent::try_from(reply)? {
23/// PubsubEvent::Message { channel, payload } => {
24/// assert_eq!((channel, payload), (b"news".to_vec(), b"hi".to_vec()));
25/// }
26/// other => panic!("unexpected {other:?}"),
27/// }
28/// # Ok::<(), Box<dyn std::error::Error>>(())
29/// ```
30#[non_exhaustive]
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub enum PubsubEvent {
33 /// `SUBSCRIBE` ack per channel.
34 ///
35 /// ```
36 /// use kevy_resp::{PubsubEvent, parse_reply};
37 /// let (reply, _) = parse_reply(b"*3\r\n$9\r\nsubscribe\r\n$4\r\nnews\r\n:1\r\n")?.expect("complete frame");
38 /// let event = PubsubEvent::try_from(reply)?;
39 /// assert_eq!(event, PubsubEvent::Subscribe { channel: b"news".to_vec(), count: 1 });
40 /// # Ok::<(), Box<dyn std::error::Error>>(())
41 /// ```
42 Subscribe {
43 /// Channel that was just subscribed.
44 ///
45 /// ```
46 /// use kevy_resp::{PubsubEvent, parse_reply};
47 /// let (reply, _) = parse_reply(b"*3\r\n$9\r\nsubscribe\r\n$4\r\nnews\r\n:1\r\n")?.expect("complete frame");
48 /// let event = PubsubEvent::try_from(reply)?;
49 /// let PubsubEvent::Subscribe { channel, .. } = event else { panic!("not subscribe") };
50 /// assert_eq!(channel, b"news");
51 /// # Ok::<(), Box<dyn std::error::Error>>(())
52 /// ```
53 channel: Vec<u8>,
54 /// Total channels + patterns subscribed.
55 ///
56 /// ```
57 /// use kevy_resp::{PubsubEvent, parse_reply};
58 /// let (reply, _) = parse_reply(b"*3\r\n$9\r\nsubscribe\r\n$4\r\nnews\r\n:1\r\n")?.expect("complete frame");
59 /// let event = PubsubEvent::try_from(reply)?;
60 /// let PubsubEvent::Subscribe { count, .. } = event else { panic!("not subscribe") };
61 /// assert_eq!(count, 1);
62 /// # Ok::<(), Box<dyn std::error::Error>>(())
63 /// ```
64 count: i64,
65 },
66 /// `PSUBSCRIBE` ack per pattern.
67 ///
68 /// ```
69 /// use kevy_resp::{PubsubEvent, parse_reply};
70 /// let (reply, _) = parse_reply(b"*3\r\n$10\r\npsubscribe\r\n$3\r\nn.*\r\n:2\r\n")?.expect("complete frame");
71 /// let event = PubsubEvent::try_from(reply)?;
72 /// assert_eq!(event, PubsubEvent::Psubscribe { pattern: b"n.*".to_vec(), count: 2 });
73 /// # Ok::<(), Box<dyn std::error::Error>>(())
74 /// ```
75 Psubscribe {
76 /// Pattern that was just subscribed.
77 ///
78 /// ```
79 /// use kevy_resp::{PubsubEvent, parse_reply};
80 /// let (reply, _) = parse_reply(b"*3\r\n$10\r\npsubscribe\r\n$3\r\nn.*\r\n:2\r\n")?.expect("complete frame");
81 /// let event = PubsubEvent::try_from(reply)?;
82 /// let PubsubEvent::Psubscribe { pattern, .. } = event else { panic!("not psubscribe") };
83 /// assert_eq!(pattern, b"n.*");
84 /// # Ok::<(), Box<dyn std::error::Error>>(())
85 /// ```
86 pattern: Vec<u8>,
87 /// Total channels + patterns subscribed.
88 ///
89 /// ```
90 /// use kevy_resp::{PubsubEvent, parse_reply};
91 /// let (reply, _) = parse_reply(b"*3\r\n$10\r\npsubscribe\r\n$3\r\nn.*\r\n:2\r\n")?.expect("complete frame");
92 /// let event = PubsubEvent::try_from(reply)?;
93 /// // the connection now holds one channel and this pattern
94 /// let PubsubEvent::Psubscribe { count, .. } = event else { panic!("not psubscribe") };
95 /// assert_eq!(count, 2);
96 /// # Ok::<(), Box<dyn std::error::Error>>(())
97 /// ```
98 count: i64,
99 },
100 /// `UNSUBSCRIBE` ack.
101 ///
102 /// ```
103 /// use kevy_resp::{PubsubEvent, parse_reply};
104 /// let (reply, _) = parse_reply(b"*3\r\n$11\r\nunsubscribe\r\n$4\r\nnews\r\n:0\r\n")?.expect("complete frame");
105 /// let event = PubsubEvent::try_from(reply)?;
106 /// assert_eq!(event, PubsubEvent::Unsubscribe { channel: Some(b"news".to_vec()), count: 0 });
107 /// # Ok::<(), Box<dyn std::error::Error>>(())
108 /// ```
109 Unsubscribe {
110 /// Channel just unsubscribed (`None` for "all"/"none" nil bulk).
111 ///
112 /// ```
113 /// use kevy_resp::{PubsubEvent, parse_reply};
114 /// let (reply, _) = parse_reply(b"*3\r\n$11\r\nunsubscribe\r\n$-1\r\n:0\r\n")?.expect("complete frame");
115 /// let event = PubsubEvent::try_from(reply)?;
116 /// // "unsubscribed from everything" arrives as a nil bulk
117 /// let PubsubEvent::Unsubscribe { channel, .. } = event else { panic!("not unsubscribe") };
118 /// assert_eq!(channel, None);
119 /// # Ok::<(), Box<dyn std::error::Error>>(())
120 /// ```
121 channel: Option<Vec<u8>>,
122 /// Total channels + patterns still subscribed.
123 ///
124 /// ```
125 /// use kevy_resp::{PubsubEvent, parse_reply};
126 /// let (reply, _) = parse_reply(b"*3\r\n$11\r\nunsubscribe\r\n$4\r\nnews\r\n:0\r\n")?.expect("complete frame");
127 /// let event = PubsubEvent::try_from(reply)?;
128 /// let PubsubEvent::Unsubscribe { count, .. } = event else { panic!("not unsubscribe") };
129 /// assert_eq!(count, 0);
130 /// # Ok::<(), Box<dyn std::error::Error>>(())
131 /// ```
132 count: i64,
133 },
134 /// `PUNSUBSCRIBE` ack.
135 ///
136 /// ```
137 /// use kevy_resp::{PubsubEvent, parse_reply};
138 /// let (reply, _) = parse_reply(b"*3\r\n$12\r\npunsubscribe\r\n$3\r\nn.*\r\n:1\r\n")?.expect("complete frame");
139 /// let event = PubsubEvent::try_from(reply)?;
140 /// assert_eq!(event, PubsubEvent::Punsubscribe { pattern: Some(b"n.*".to_vec()), count: 1 });
141 /// # Ok::<(), Box<dyn std::error::Error>>(())
142 /// ```
143 Punsubscribe {
144 /// Pattern just unsubscribed (`None` for "all"/"none" nil bulk).
145 ///
146 /// ```
147 /// use kevy_resp::{PubsubEvent, parse_reply};
148 /// let (reply, _) = parse_reply(b"*3\r\n$12\r\npunsubscribe\r\n$-1\r\n:0\r\n")?.expect("complete frame");
149 /// let event = PubsubEvent::try_from(reply)?;
150 /// let PubsubEvent::Punsubscribe { pattern, .. } = event else { panic!("not punsubscribe") };
151 /// assert_eq!(pattern, None);
152 /// # Ok::<(), Box<dyn std::error::Error>>(())
153 /// ```
154 pattern: Option<Vec<u8>>,
155 /// Total channels + patterns still subscribed.
156 ///
157 /// ```
158 /// use kevy_resp::{PubsubEvent, parse_reply};
159 /// let (reply, _) = parse_reply(b"*3\r\n$12\r\npunsubscribe\r\n$3\r\nn.*\r\n:1\r\n")?.expect("complete frame");
160 /// let event = PubsubEvent::try_from(reply)?;
161 /// // one channel subscription is still active
162 /// let PubsubEvent::Punsubscribe { count, .. } = event else { panic!("not punsubscribe") };
163 /// assert_eq!(count, 1);
164 /// # Ok::<(), Box<dyn std::error::Error>>(())
165 /// ```
166 count: i64,
167 },
168 /// Plain `PUBLISH` delivery on a subscribed channel.
169 ///
170 /// ```
171 /// use kevy_resp::{PubsubEvent, parse_reply};
172 /// // a RESP3 push frame
173 /// let (reply, _) = parse_reply(b">3\r\n$7\r\nmessage\r\n$4\r\nnews\r\n$2\r\nhi\r\n")?.expect("complete frame");
174 /// let event = PubsubEvent::try_from(reply)?;
175 /// assert_eq!(event, PubsubEvent::Message { channel: b"news".to_vec(), payload: b"hi".to_vec() });
176 /// # Ok::<(), Box<dyn std::error::Error>>(())
177 /// ```
178 Message {
179 /// Channel the publish was made to.
180 ///
181 /// ```
182 /// use kevy_resp::{PubsubEvent, parse_reply};
183 /// let (reply, _) = parse_reply(b">3\r\n$7\r\nmessage\r\n$4\r\nnews\r\n$2\r\nhi\r\n")?.expect("complete frame");
184 /// let event = PubsubEvent::try_from(reply)?;
185 /// let PubsubEvent::Message { channel, .. } = event else { panic!("not message") };
186 /// assert_eq!(channel, b"news");
187 /// # Ok::<(), Box<dyn std::error::Error>>(())
188 /// ```
189 channel: Vec<u8>,
190 /// Raw payload bytes.
191 ///
192 /// ```
193 /// use kevy_resp::{PubsubEvent, parse_reply};
194 /// let (reply, _) = parse_reply(b">3\r\n$7\r\nmessage\r\n$4\r\nnews\r\n$2\r\nhi\r\n")?.expect("complete frame");
195 /// let event = PubsubEvent::try_from(reply)?;
196 /// assert_eq!(event.into_payload(), Some(b"hi".to_vec()));
197 /// # Ok::<(), Box<dyn std::error::Error>>(())
198 /// ```
199 payload: Vec<u8>,
200 },
201 /// Pattern-match delivery.
202 ///
203 /// ```
204 /// use kevy_resp::{PubsubEvent, parse_reply};
205 /// let (reply, _) = parse_reply(b"*4\r\n$8\r\npmessage\r\n$3\r\nn.*\r\n$4\r\nnews\r\n$2\r\nhi\r\n")?.expect("complete frame");
206 /// let event = PubsubEvent::try_from(reply)?;
207 /// let expected = PubsubEvent::Pmessage {
208 /// pattern: b"n.*".to_vec(),
209 /// channel: b"news".to_vec(),
210 /// payload: b"hi".to_vec(),
211 /// };
212 /// assert_eq!(event, expected);
213 /// # Ok::<(), Box<dyn std::error::Error>>(())
214 /// ```
215 Pmessage {
216 /// Pattern the channel matched.
217 ///
218 /// ```
219 /// use kevy_resp::{PubsubEvent, parse_reply};
220 /// let (reply, _) = parse_reply(b"*4\r\n$8\r\npmessage\r\n$3\r\nn.*\r\n$4\r\nnews\r\n$2\r\nhi\r\n")?.expect("complete frame");
221 /// let event = PubsubEvent::try_from(reply)?;
222 /// let PubsubEvent::Pmessage { pattern, .. } = event else { panic!("not pmessage") };
223 /// assert_eq!(pattern, b"n.*");
224 /// # Ok::<(), Box<dyn std::error::Error>>(())
225 /// ```
226 pattern: Vec<u8>,
227 /// Channel the publish was made to.
228 ///
229 /// ```
230 /// use kevy_resp::{PubsubEvent, parse_reply};
231 /// let (reply, _) = parse_reply(b"*4\r\n$8\r\npmessage\r\n$3\r\nn.*\r\n$4\r\nnews\r\n$2\r\nhi\r\n")?.expect("complete frame");
232 /// let event = PubsubEvent::try_from(reply)?;
233 /// let PubsubEvent::Pmessage { channel, .. } = event else { panic!("not pmessage") };
234 /// assert_eq!(channel, b"news");
235 /// # Ok::<(), Box<dyn std::error::Error>>(())
236 /// ```
237 channel: Vec<u8>,
238 /// Raw payload bytes.
239 ///
240 /// ```
241 /// use kevy_resp::{PubsubEvent, parse_reply};
242 /// let (reply, _) = parse_reply(b"*4\r\n$8\r\npmessage\r\n$3\r\nn.*\r\n$4\r\nnews\r\n$2\r\nhi\r\n")?.expect("complete frame");
243 /// let event = PubsubEvent::try_from(reply)?;
244 /// let PubsubEvent::Pmessage { payload, .. } = event else { panic!("not pmessage") };
245 /// assert_eq!(payload, b"hi");
246 /// # Ok::<(), Box<dyn std::error::Error>>(())
247 /// ```
248 payload: Vec<u8>,
249 },
250}
251
252/// Turn a RESP reply into a [`PubsubEvent`]. Handles both RESP2
253/// (`*N\r\n…` arrays) and RESP3 (`>N\r\n…` push frames).
254///
255/// ```
256/// use kevy_resp::{PubsubEvent, Reply};
257///
258/// let r = Reply::Array(vec![
259/// Reply::Bulk(b"message".to_vec()),
260/// Reply::Bulk(b"news".to_vec()),
261/// Reply::Bulk(b"hi".to_vec()),
262/// ]);
263/// assert!(matches!(PubsubEvent::try_from(r)?, PubsubEvent::Message { .. }));
264/// # Ok::<(), std::io::Error>(())
265/// ```
266impl PubsubEvent {
267 /// The raw message payload, moved out of the event.
268 ///
269 /// `Some(payload)` for the two deliveries ([`Message`](Self::Message)
270 /// and [`Pmessage`](Self::Pmessage)); `None` for every control/ack
271 /// event (subscribe / unsubscribe / …), which carries no payload.
272 /// Consuming `self` lets a scalar drain hand a push subscriber just the
273 /// bytes with no extra copy. The channel and the message-vs-pmessage
274 /// distinction are dropped; a caller that needs either keeps matching
275 /// on the event.
276 ///
277 /// ```
278 /// use kevy_resp::PubsubEvent;
279 ///
280 /// let m = PubsubEvent::Message { channel: b"c".to_vec(), payload: b"p".to_vec() };
281 /// assert_eq!(m.into_payload(), Some(b"p".to_vec()));
282 /// let ack = PubsubEvent::Subscribe { channel: b"c".to_vec(), count: 1 };
283 /// assert_eq!(ack.into_payload(), None);
284 /// ```
285 #[inline]
286 #[must_use]
287 pub fn into_payload(self) -> Option<Vec<u8>> {
288 match self {
289 PubsubEvent::Message { payload, .. } | PubsubEvent::Pmessage { payload, .. } => {
290 Some(payload)
291 }
292 _ => None,
293 }
294 }
295}
296
297impl TryFrom<Reply> for PubsubEvent {
298 type Error = io::Error;
299
300 // LOC-WAIVER: data-driven pubsub-kind match table — one flat frame-destructure arm per kind.
301 fn try_from(reply: Reply) -> io::Result<PubsubEvent> {
302 let items = match reply {
303 Reply::Array(v) | Reply::Push(v) => v,
304 Reply::Error(e) => {
305 return Err(io::Error::other(String::from_utf8_lossy(&e).into_owned()));
306 }
307 other => {
308 return Err(invalid(format!("pubsub: expected array/push, got {}", shape(&other))));
309 }
310 };
311
312 let mut it = items.into_iter();
313 let kind = take_bulk(it.next().ok_or_else(|| invalid("pubsub: empty frame"))?, "kind")?;
314
315 match kind.as_slice() {
316 b"subscribe" => {
317 let channel = take_bulk(
318 it.next().ok_or_else(|| invalid("subscribe: missing channel"))?,
319 "channel",
320 )?;
321 let count = take_int(
322 it.next().ok_or_else(|| invalid("subscribe: missing count"))?,
323 "count",
324 )?;
325 Ok(PubsubEvent::Subscribe { channel, count })
326 }
327 b"psubscribe" => {
328 let pattern = take_bulk(
329 it.next().ok_or_else(|| invalid("psubscribe: missing pattern"))?,
330 "pattern",
331 )?;
332 let count = take_int(
333 it.next().ok_or_else(|| invalid("psubscribe: missing count"))?,
334 "count",
335 )?;
336 Ok(PubsubEvent::Psubscribe { pattern, count })
337 }
338 b"unsubscribe" => {
339 let channel = take_bulk_or_nil(
340 it.next().ok_or_else(|| invalid("unsubscribe: missing channel"))?,
341 "channel",
342 )?;
343 let count = take_int(
344 it.next().ok_or_else(|| invalid("unsubscribe: missing count"))?,
345 "count",
346 )?;
347 Ok(PubsubEvent::Unsubscribe { channel, count })
348 }
349 b"punsubscribe" => {
350 let pattern = take_bulk_or_nil(
351 it.next().ok_or_else(|| invalid("punsubscribe: missing pattern"))?,
352 "pattern",
353 )?;
354 let count = take_int(
355 it.next().ok_or_else(|| invalid("punsubscribe: missing count"))?,
356 "count",
357 )?;
358 Ok(PubsubEvent::Punsubscribe { pattern, count })
359 }
360 b"message" => {
361 let channel = take_bulk(
362 it.next().ok_or_else(|| invalid("message: missing channel"))?,
363 "channel",
364 )?;
365 let payload = take_bulk(
366 it.next().ok_or_else(|| invalid("message: missing payload"))?,
367 "payload",
368 )?;
369 Ok(PubsubEvent::Message { channel, payload })
370 }
371 b"pmessage" => {
372 let pattern = take_bulk(
373 it.next().ok_or_else(|| invalid("pmessage: missing pattern"))?,
374 "pattern",
375 )?;
376 let channel = take_bulk(
377 it.next().ok_or_else(|| invalid("pmessage: missing channel"))?,
378 "channel",
379 )?;
380 let payload = take_bulk(
381 it.next().ok_or_else(|| invalid("pmessage: missing payload"))?,
382 "payload",
383 )?;
384 Ok(PubsubEvent::Pmessage { pattern, channel, payload })
385 }
386 other => {
387 Err(invalid(format!("unknown pubsub kind: {}", String::from_utf8_lossy(other))))
388 }
389 }
390 }
391}
392
393fn take_bulk(r: Reply, field: &str) -> io::Result<Vec<u8>> {
394 match r {
395 Reply::Bulk(v) | Reply::Simple(v) => Ok(v),
396 other => {
397 Err(invalid(format!("pubsub field {field}: expected bulk, got {}", shape(&other))))
398 }
399 }
400}
401
402fn take_bulk_or_nil(r: Reply, field: &str) -> io::Result<Option<Vec<u8>>> {
403 match r {
404 Reply::Bulk(v) | Reply::Simple(v) => Ok(Some(v)),
405 Reply::Nil | Reply::Null => Ok(None),
406 other => {
407 Err(invalid(format!("pubsub field {field}: expected bulk/nil, got {}", shape(&other))))
408 }
409 }
410}
411
412fn take_int(r: Reply, field: &str) -> io::Result<i64> {
413 match r {
414 Reply::Int(n) => Ok(n),
415 other => Err(invalid(format!("pubsub field {field}: expected int, got {}", shape(&other)))),
416 }
417}
418
419fn shape(r: &Reply) -> &'static str {
420 match r {
421 Reply::Simple(_) => "simple",
422 Reply::Error(_) => "error",
423 Reply::Int(_) => "int",
424 Reply::Bulk(_) => "bulk",
425 Reply::Nil | Reply::Null => "nil",
426 Reply::Array(_) => "array",
427 Reply::Map(_) => "map",
428 Reply::Set(_) => "set",
429 Reply::Double(_) => "double",
430 Reply::Boolean(_) => "boolean",
431 Reply::Verbatim { .. } => "verbatim",
432 Reply::BigNumber(_) => "bignumber",
433 Reply::Push(_) => "push",
434 Reply::BlobError(_) => "bloberror",
435 }
436}
437
438fn invalid(msg: impl Into<String>) -> io::Error {
439 io::Error::new(io::ErrorKind::InvalidData, msg.into())
440}
441
442#[cfg(test)]
443#[path = "pubsub_event_tests.rs"]
444mod tests;