Struct RedisStreams 

pub struct RedisStreams { /* private fields */ }

Redis Streams client for event-driven architectures

Wraps Redis ConnectionManager to provide high-level streaming operations.

Implementations

impl RedisStreams

pub async fn new(redis_url: &str) -> Result<Self>

Create a new RedisStreams instance

Arguments

• redis_url - Redis connection URL (e.g., <redis://127.0.0.1:6379>)

Example

let streams = RedisStreams::new("redis://127.0.0.1:6379").await?;

Errors

Returns an error if Redis connection fails.

pub async fn stream_add( &self, stream_key: &str, fields: Vec<(String, String)>, maxlen: Option<usize>, ) -> Result<String>

Publish data to Redis Stream using XADD

Arguments

• stream_key - Name of the Redis Stream (e.g., “events_stream”)
• fields - Vec of field-value pairs to add to the stream
• maxlen - Optional maximum length for stream trimming (uses MAXLEN ~ for approximate trimming)

Returns

The stream entry ID generated by Redis (e.g., “1234567890123-0”)

Example

let event_id = streams.stream_add(
    "user_events",
    vec![
        ("user_id".to_string(), "42".to_string()),
        ("action".to_string(), "purchase".to_string()),
        ("amount".to_string(), "99.99".to_string()),
    ],
    Some(10000)  // Keep max 10k events
).await?;
println!("Event ID: {}", event_id);

Errors

Returns an error if the XADD command fails.

pub async fn stream_read_latest( &self, stream_key: &str, count: usize, ) -> Result<Vec<StreamEntry>>

Read latest entries from Redis Stream using XREVRANGE

Arguments

• stream_key - Name of the Redis Stream
• count - Number of latest entries to retrieve

Returns

Vector of (entry_id, fields) tuples, ordered from newest to oldest

Example

let latest_10 = streams.stream_read_latest("events", 10).await?;
for (id, fields) in latest_10 {
    println!("Event {}: {:?}", id, fields);
}

Errors

Returns an error if the XREVRANGE command fails.

pub async fn stream_read( &self, stream_key: &str, last_id: &str, count: usize, block_ms: Option<usize>, ) -> Result<Vec<StreamEntry>>

Read entries from Redis Stream using XREAD (blocking or non-blocking)

Arguments

• stream_key - Name of the Redis Stream
• last_id - Last entry ID seen (use “0” to read from beginning, “$” for only new entries)
• count - Maximum number of entries to retrieve
• block_ms - Optional blocking timeout in milliseconds (None for non-blocking)

Returns

Vector of (entry_id, fields) tuples

Example

// Non-blocking: read new entries since last seen ID
let entries = streams.stream_read("events", "1234567890-0", 100, None).await?;

// Blocking: wait up to 5 seconds for new entries
let new_entries = streams.stream_read("events", "$", 100, Some(5000)).await?;

Errors

Returns an error if the XREAD command fails.

Trait Implementations

impl Clone for RedisStreams

fn clone(&self) -> RedisStreams

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl StreamingBackend for RedisStreams

Implement StreamingBackend trait for RedisStreams

This enables RedisStreams to be used as a pluggable streaming backend.

fn stream_add<'life0, 'life1, 'async_trait>( &'life0 self, stream_key: &'life1 str, fields: Vec<(String, String)>, maxlen: Option<usize>, ) -> Pin<Box<dyn Future<Output = Result<String>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Add an entry to a stream

fn stream_read_latest<'life0, 'life1, 'async_trait>( &'life0 self, stream_key: &'life1 str, count: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, Vec<(String, String)>)>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Read the latest N entries from a stream (newest first)

fn stream_read<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, stream_key: &'life1 str, last_id: &'life2 str, count: usize, block_ms: Option<usize>, ) -> Pin<Box<dyn Future<Output = Result<Vec<(String, Vec<(String, String)>)>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Read entries from a stream with optional blocking