Redis Streams Backend
Relevant source files
The following files were used as context for generating this wiki page:
The Redis Streams backend for SeaStreamer provides a high-level, asynchronous API for interacting with Redis Streams. It maps the abstract sea-streamer traits to Redis-specific commands like XADD, XREAD, and XACK, while abstracting away the complexities of Redis connection management and stream semantics. The backend is designed for high performance, utilizing a mutex-free implementation and achieving throughput in the realm of 100k messages per second through pipelined operations.
Sources: sea-streamer-redis/src/lib.rs:1-12, sea-streamer-redis/README.md:3-12
Architecture and Concurrency Model
SeaStreamer Redis achieves high concurrency by separating the read and write loops from the main processing loop. This message-passing architecture ensures that while the application is processing a message, the backend can simultaneously fetch the next batch or flush pending writes to Redis.
The diagram illustrates how SeaStreamer decouples I/O operations from message processing to maximize throughput.
Sources: sea-streamer-redis/README.md:27-46, sea-streamer-redis/src/lib.rs:27-45
Key Backend Components
The backend is structured around several core components that implement the standard SeaStreamer traits.
RedisStreamer
The entry point for the backend. It handles connection establishment, credential management, and the creation of producers and consumers. It also supports Redis Cluster configurations.
Sources: sea-streamer-redis/src/streamer.rs:11-15
RedisProducer
Handles sending messages to Redis Streams. It supports pipelining XADD commands for efficiency and provides functionality for stream trimming (XTRIM).
Sources: sea-streamer-redis/src/producer.rs:1-10, CHANGELOG.md:21-25
RedisConsumer
Facilitates reading from streams. It supports various ConsumerMode settings, including RealTime, Resumable (with auto-ack/commit), and LoadBalanced modes.
Sources: sea-streamer-redis/README.md:21-26, sea-streamer-redis/src/consumer/mod.rs:1-5
RedisManager
A specialized component for administrative tasks, such as managing stream groups and trimming streams.
Sources: sea-streamer-redis/src/manager.rs:1-5, sea-streamer-redis/src/streamer.rs:90-96
Configuration Options
The backend provides several configuration structures to tune connection behavior and stream interactions.
| Option | Type | Default | Description |
|---|---|---|---|
db | u32 | 0 | The Redis database index to use. |
timeout | Duration | 60s | Network request timeout. |
enable_cluster | bool | false | Enables support for Redis Cluster. |
timestamp_format | TimestampFormat | Millis | Precision of timestamps (Millis or Nanos). |
message_field | &str | "msg" | The Redis hash field name for the payload. |
Sources: sea-streamer-redis/src/streamer.rs:17-26, sea-streamer-redis/src/lib.rs:88-90
Connection Sequence
Connecting to a Redis backend involves normalizing the URI and initializing the internal RedisCluster (which manages one or more node connections).
Redis vs. Kafka Semantic Differences
The backend implementation accounts for fundamental differences between Redis Streams and Apache Kafka:
| Feature | Redis Streams | Apache Kafka |
|---|---|---|
| Sequence Numbers | Non-contiguous (e.g., 162...-0) | Contiguous integers |
| Dispatching | First-ask-first-served in groups | 1-to-1 Consumer to Shard mapping |
| Acknowledgment | Required per message (XACK) | Read-up-to (Cumulative ACK) |
Sources: sea-streamer-redis/README.md:14-20, sea-streamer-redis/src/lib.rs:14-25
Technical Implementation Details
Message Representation
Messages in Redis Streams are stored as hashes. By default, SeaStreamer uses a single field named msg to store the payload, but this is configurable via RedisConnectOptions.
pub const MSG: &str = "msg"; // Default field
// In sea-streamer-redis/src/streamer.rs
pub fn set_message_field(&mut self, field: &'static str) -> &mut Self {
self.message_field = MessageField(field);
self
}
Sources: sea-streamer-redis/src/lib.rs:93, sea-streamer-redis/src/streamer.rs:159-163
Sharding
Basic stream sharding is supported by splitting a stream into multiple sub-streams. However, currently, the implementation defaults most operations to ShardId zero if not explicitly managed.
Sources: sea-streamer-redis/src/lib.rs:25, sea-streamer-redis/README.md:26
Summary
The Redis Streams Backend provides a high-performance, asynchronous wrapper around Redis's native streaming capabilities. By bridging the gap between Redis's unique semantics and the SeaStreamer generic interface, it allows developers to build backend-agnostic stream processors that benefit from Redis's low latency and high throughput.