redis_oxide/
streams.rs

1//! Redis Streams support for event sourcing and messaging
2//!
3//! Redis Streams provide a powerful data structure for handling time-series data,
4//! event sourcing, and message queuing. This module provides a high-level API
5//! for working with Redis Streams.
6//!
7//! # Examples
8//!
9//! ## Basic Stream Operations
10//!
11//! ```no_run
12//! use redis_oxide::{Client, ConnectionConfig};
13//! use std::collections::HashMap;
14//!
15//! # #[tokio::main]
16//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
17//! let config = ConnectionConfig::new("redis://localhost:6379");
18//! let client = Client::connect(config).await?;
19//!
20//! // Add entries to a stream
21//! let mut fields = HashMap::new();
22//! fields.insert("user_id".to_string(), "123".to_string());
23//! fields.insert("action".to_string(), "login".to_string());
24//!
25//! let entry_id = client.xadd("events", "*", fields).await?;
26//! println!("Added entry: {}", entry_id);
27//!
28//! // Read from stream
29//! let entries = client.xrange("events", "-", "+", Some(10)).await?;
30//! for entry in entries {
31//!     println!("Entry {}: {:?}", entry.id, entry.fields);
32//! }
33//! # Ok(())
34//! # }
35//! ```
36//!
37//! ## Consumer Groups
38//!
39//! ```no_run
40//! use redis_oxide::{Client, ConnectionConfig};
41//!
42//! # #[tokio::main]
43//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
44//! let config = ConnectionConfig::new("redis://localhost:6379");
45//! let client = Client::connect(config).await?;
46//!
47//! // Create a consumer group
48//! client.xgroup_create("events", "processors", "$", true).await?;
49//!
50//! // Read from the group
51//! let messages = client.xreadgroup(
52//!     "processors",
53//!     "worker-1",
54//!     vec![("events".to_string(), ">".to_string())],
55//!     Some(1),
56//!     Some(std::time::Duration::from_secs(1))
57//! ).await?;
58//!
59//! for (stream, entries) in messages {
60//!     for entry in entries {
61//!         println!("Processing {}: {:?}", entry.id, entry.fields);
62//!         // Process the message...
63//!         
64//!         // Acknowledge the message
65//!         client.xack("events", "processors", vec![entry.id]).await?;
66//!     }
67//! }
68//! # Ok(())
69//! # }
70//! ```
71
72use crate::core::{
73    error::{RedisError, RedisResult},
74    value::RespValue,
75};
76use std::collections::HashMap;
77use std::time::Duration;
78
79/// Represents a single entry in a Redis Stream
80#[derive(Debug, Clone)]
81pub struct StreamEntry {
82    /// The unique ID of the entry (e.g., "1234567890123-0")
83    pub id: String,
84    /// The field-value pairs of the entry
85    pub fields: HashMap<String, String>,
86}
87
88impl StreamEntry {
89    /// Create a new stream entry
90    pub fn new(id: String, fields: HashMap<String, String>) -> Self {
91        Self { id, fields }
92    }
93
94    /// Get a field value by name
95    #[must_use]
96    pub fn get_field(&self, field: &str) -> Option<&String> {
97        self.fields.get(field)
98    }
99
100    /// Check if the entry has a specific field
101    #[must_use]
102    pub fn has_field(&self, field: &str) -> bool {
103        self.fields.contains_key(field)
104    }
105
106    /// Get the timestamp part of the entry ID
107    ///
108    /// # Examples
109    ///
110    /// ```
111    /// use redis_oxide::StreamEntry;
112    /// use std::collections::HashMap;
113    ///
114    /// let entry = StreamEntry::new("1234567890123-0".to_string(), HashMap::new());
115    /// assert_eq!(entry.timestamp(), Some(1234567890123));
116    /// ```
117    #[must_use]
118    pub fn timestamp(&self) -> Option<u64> {
119        self.id.split('-').next()?.parse().ok()
120    }
121
122    /// Get the sequence number part of the entry ID
123    ///
124    /// # Examples
125    ///
126    /// ```
127    /// use redis_oxide::StreamEntry;
128    /// use std::collections::HashMap;
129    ///
130    /// let entry = StreamEntry::new("1234567890123-5".to_string(), HashMap::new());
131    /// assert_eq!(entry.sequence(), Some(5));
132    /// ```
133    #[must_use]
134    pub fn sequence(&self) -> Option<u64> {
135        self.id.split('-').nth(1)?.parse().ok()
136    }
137}
138
139/// Information about a Redis Stream
140#[derive(Debug, Clone)]
141pub struct StreamInfo {
142    /// The length of the stream (number of entries)
143    pub length: u64,
144    /// The number of consumer groups
145    pub groups: u64,
146    /// The ID of the first entry
147    pub first_entry: Option<String>,
148    /// The ID of the last entry
149    pub last_entry: Option<String>,
150    /// The last generated ID
151    pub last_generated_id: String,
152}
153
154/// Information about a consumer group
155#[derive(Debug, Clone)]
156pub struct ConsumerGroupInfo {
157    /// The name of the consumer group
158    pub name: String,
159    /// The number of consumers in the group
160    pub consumers: u64,
161    /// The number of pending messages
162    pub pending: u64,
163    /// The ID of the last delivered message
164    pub last_delivered_id: String,
165}
166
167/// Information about a consumer in a group
168#[derive(Debug, Clone)]
169pub struct ConsumerInfo {
170    /// The name of the consumer
171    pub name: String,
172    /// The number of pending messages for this consumer
173    pub pending: u64,
174    /// The idle time in milliseconds
175    pub idle: u64,
176}
177
178/// Pending message information
179#[derive(Debug, Clone)]
180pub struct PendingMessage {
181    /// The message ID
182    pub id: String,
183    /// The consumer name
184    pub consumer: String,
185    /// The idle time in milliseconds
186    pub idle_time: u64,
187    /// The delivery count
188    pub delivery_count: u64,
189}
190
191/// Stream range options for XRANGE and XREVRANGE
192#[derive(Debug, Clone)]
193pub struct StreamRange {
194    /// Start ID (inclusive)
195    pub start: String,
196    /// End ID (inclusive)
197    pub end: String,
198    /// Maximum number of entries to return
199    pub count: Option<u64>,
200}
201
202impl StreamRange {
203    /// Create a new stream range
204    pub fn new(start: impl Into<String>, end: impl Into<String>) -> Self {
205        Self {
206            start: start.into(),
207            end: end.into(),
208            count: None,
209        }
210    }
211
212    /// Set the maximum number of entries to return
213    pub fn with_count(mut self, count: u64) -> Self {
214        self.count = Some(count);
215        self
216    }
217
218    /// Create a range for all entries
219    pub fn all() -> Self {
220        Self::new("-", "+")
221    }
222
223    /// Create a range from a specific ID to the end
224    pub fn from(start: impl Into<String>) -> Self {
225        Self::new(start, "+")
226    }
227
228    /// Create a range from the beginning to a specific ID
229    pub fn to(end: impl Into<String>) -> Self {
230        Self::new("-", end)
231    }
232}
233
234/// Options for XREAD and XREADGROUP commands
235#[derive(Debug, Clone)]
236pub struct ReadOptions {
237    /// Maximum number of entries per stream
238    pub count: Option<u64>,
239    /// Block timeout in milliseconds (None for non-blocking)
240    pub block: Option<Duration>,
241}
242
243impl ReadOptions {
244    /// Create new read options
245    #[must_use]
246    pub fn new() -> Self {
247        Self {
248            count: None,
249            block: None,
250        }
251    }
252
253    /// Set the maximum number of entries per stream
254    pub fn with_count(mut self, count: u64) -> Self {
255        self.count = Some(count);
256        self
257    }
258
259    /// Set the block timeout
260    pub fn with_block(mut self, timeout: Duration) -> Self {
261        self.block = Some(timeout);
262        self
263    }
264
265    /// Create blocking read options
266    pub fn blocking(timeout: Duration) -> Self {
267        Self::new().with_block(timeout)
268    }
269
270    /// Create non-blocking read options with count limit
271    pub fn non_blocking(count: u64) -> Self {
272        Self::new().with_count(count)
273    }
274}
275
276impl Default for ReadOptions {
277    fn default() -> Self {
278        Self::new()
279    }
280}
281
282/// Parse stream entries from Redis response
283pub fn parse_stream_entries(response: RespValue) -> RedisResult<Vec<StreamEntry>> {
284    match response {
285        RespValue::Array(items) => {
286            let mut entries = Vec::new();
287
288            for item in items {
289                match item {
290                    RespValue::Array(entry_data) if entry_data.len() == 2 => {
291                        let id = entry_data[0].as_string()?;
292
293                        match &entry_data[1] {
294                            RespValue::Array(field_values) => {
295                                let mut fields = HashMap::new();
296
297                                // Fields are stored as [field1, value1, field2, value2, ...]
298                                for chunk in field_values.chunks(2) {
299                                    if chunk.len() == 2 {
300                                        let field = chunk[0].as_string()?;
301                                        let value = chunk[1].as_string()?;
302                                        fields.insert(field, value);
303                                    }
304                                }
305
306                                entries.push(StreamEntry::new(id, fields));
307                            }
308                            _ => {
309                                return Err(RedisError::Type(format!(
310                                    "Invalid stream entry field format: {:?}",
311                                    entry_data[1]
312                                )))
313                            }
314                        }
315                    }
316                    _ => {
317                        return Err(RedisError::Type(format!(
318                            "Invalid stream entry format: {:?}",
319                            item
320                        )))
321                    }
322                }
323            }
324
325            Ok(entries)
326        }
327        _ => Err(RedisError::Type(format!(
328            "Expected array for stream entries, got: {:?}",
329            response
330        ))),
331    }
332}
333
334/// Parse XREAD/XREADGROUP response
335pub fn parse_xread_response(response: RespValue) -> RedisResult<HashMap<String, Vec<StreamEntry>>> {
336    match response {
337        RespValue::Array(streams) => {
338            let mut result = HashMap::new();
339
340            for stream in streams {
341                match stream {
342                    RespValue::Array(stream_data) if stream_data.len() == 2 => {
343                        let stream_name = stream_data[0].as_string()?;
344                        let entries = parse_stream_entries(stream_data[1].clone())?;
345                        result.insert(stream_name, entries);
346                    }
347                    _ => {
348                        return Err(RedisError::Type(format!(
349                            "Invalid XREAD response format: {:?}",
350                            stream
351                        )))
352                    }
353                }
354            }
355
356            Ok(result)
357        }
358        RespValue::Null => Ok(HashMap::new()), // No new entries
359        _ => Err(RedisError::Type(format!(
360            "Expected array or null for XREAD response, got: {:?}",
361            response
362        ))),
363    }
364}
365
366/// Parse stream info from XINFO STREAM response
367pub fn parse_stream_info(response: RespValue) -> RedisResult<StreamInfo> {
368    match response {
369        RespValue::Array(items) => {
370            let mut length = 0;
371            let mut groups = 0;
372            let mut first_entry = None;
373            let mut last_entry = None;
374            let mut last_generated_id = String::new();
375
376            // Parse key-value pairs
377            for chunk in items.chunks(2) {
378                if chunk.len() == 2 {
379                    let key = chunk[0].as_string()?;
380                    match key.as_str() {
381                        "length" => length = chunk[1].as_int()? as u64,
382                        "groups" => groups = chunk[1].as_int()? as u64,
383                        "first-entry" => {
384                            first_entry = parse_stream_info_entry_id(&chunk[1])?;
385                        }
386                        "last-entry" => {
387                            last_entry = parse_stream_info_entry_id(&chunk[1])?;
388                        }
389                        "last-generated-id" => last_generated_id = chunk[1].as_string()?,
390                        _ => {} // Ignore unknown fields
391                    }
392                }
393            }
394
395            Ok(StreamInfo {
396                length,
397                groups,
398                first_entry,
399                last_entry,
400                last_generated_id,
401            })
402        }
403        _ => Err(RedisError::Type(format!(
404            "Expected array for stream info, got: {:?}",
405            response
406        ))),
407    }
408}
409
410fn parse_stream_info_entry_id(value: &RespValue) -> RedisResult<Option<String>> {
411    match value {
412        RespValue::Array(entry) => entry.first().map(RespValue::as_string).transpose(),
413        _ => Ok(None),
414    }
415}
416
417#[cfg(test)]
418mod tests {
419    use super::*;
420
421    #[test]
422    fn test_stream_entry_creation() {
423        let mut fields = HashMap::new();
424        fields.insert("user".to_string(), "alice".to_string());
425        fields.insert("action".to_string(), "login".to_string());
426
427        let entry = StreamEntry::new("1234567890123-0".to_string(), fields.clone());
428
429        assert_eq!(entry.id, "1234567890123-0");
430        assert_eq!(entry.fields, fields);
431        assert_eq!(entry.get_field("user"), Some(&"alice".to_string()));
432        assert!(entry.has_field("action"));
433        assert!(!entry.has_field("nonexistent"));
434    }
435
436    #[test]
437    fn test_stream_entry_timestamp_parsing() {
438        let entry = StreamEntry::new("1234567890123-5".to_string(), HashMap::new());
439
440        assert_eq!(entry.timestamp(), Some(1_234_567_890_123));
441        assert_eq!(entry.sequence(), Some(5));
442    }
443
444    #[test]
445    fn test_stream_entry_invalid_id() {
446        let entry = StreamEntry::new("invalid-id".to_string(), HashMap::new());
447
448        assert_eq!(entry.timestamp(), None);
449        assert_eq!(entry.sequence(), None);
450    }
451
452    #[test]
453    fn test_stream_range_creation() {
454        let range = StreamRange::new("1000", "2000").with_count(10);
455
456        assert_eq!(range.start, "1000");
457        assert_eq!(range.end, "2000");
458        assert_eq!(range.count, Some(10));
459    }
460
461    #[test]
462    fn test_stream_range_presets() {
463        let all = StreamRange::all();
464        assert_eq!(all.start, "-");
465        assert_eq!(all.end, "+");
466
467        let from = StreamRange::from("1000");
468        assert_eq!(from.start, "1000");
469        assert_eq!(from.end, "+");
470
471        let to = StreamRange::to("2000");
472        assert_eq!(to.start, "-");
473        assert_eq!(to.end, "2000");
474    }
475
476    #[test]
477    fn test_read_options() {
478        let options = ReadOptions::new()
479            .with_count(5)
480            .with_block(Duration::from_secs(1));
481
482        assert_eq!(options.count, Some(5));
483        assert_eq!(options.block, Some(Duration::from_secs(1)));
484
485        let blocking = ReadOptions::blocking(Duration::from_millis(500));
486        assert_eq!(blocking.block, Some(Duration::from_millis(500)));
487
488        let non_blocking = ReadOptions::non_blocking(10);
489        assert_eq!(non_blocking.count, Some(10));
490        assert_eq!(non_blocking.block, None);
491    }
492
493    #[test]
494    fn test_parse_stream_entries() {
495        let response = RespValue::Array(vec![
496            RespValue::Array(vec![
497                RespValue::from("1234567890123-0"),
498                RespValue::Array(vec![
499                    RespValue::from("user"),
500                    RespValue::from("alice"),
501                    RespValue::from("action"),
502                    RespValue::from("login"),
503                ]),
504            ]),
505            RespValue::Array(vec![
506                RespValue::from("1234567890124-0"),
507                RespValue::Array(vec![
508                    RespValue::from("user"),
509                    RespValue::from("bob"),
510                    RespValue::from("action"),
511                    RespValue::from("logout"),
512                ]),
513            ]),
514        ]);
515
516        let entries = parse_stream_entries(response).unwrap();
517        assert_eq!(entries.len(), 2);
518
519        assert_eq!(entries[0].id, "1234567890123-0");
520        assert_eq!(entries[0].get_field("user"), Some(&"alice".to_string()));
521        assert_eq!(entries[0].get_field("action"), Some(&"login".to_string()));
522
523        assert_eq!(entries[1].id, "1234567890124-0");
524        assert_eq!(entries[1].get_field("user"), Some(&"bob".to_string()));
525        assert_eq!(entries[1].get_field("action"), Some(&"logout".to_string()));
526    }
527
528    #[test]
529    fn test_parse_xread_response() {
530        let response = RespValue::Array(vec![RespValue::Array(vec![
531            RespValue::from("stream1"),
532            RespValue::Array(vec![RespValue::Array(vec![
533                RespValue::from("1234567890123-0"),
534                RespValue::Array(vec![RespValue::from("field1"), RespValue::from("value1")]),
535            ])]),
536        ])]);
537
538        let result = parse_xread_response(response).unwrap();
539        assert_eq!(result.len(), 1);
540        assert!(result.contains_key("stream1"));
541
542        let entries = &result["stream1"];
543        assert_eq!(entries.len(), 1);
544        assert_eq!(entries[0].id, "1234567890123-0");
545        assert_eq!(entries[0].get_field("field1"), Some(&"value1".to_string()));
546    }
547
548    #[test]
549    fn test_parse_xread_response_null() {
550        let response = RespValue::Null;
551        let result = parse_xread_response(response).unwrap();
552        assert!(result.is_empty());
553    }
554}
redis_oxide/streams.rs

redis_oxide/
streams.rs
