Struct RedisStream 

pub struct RedisStream { /* private fields */ }

Describes one Redis Streams subscription against crate::RedisBroker.

Examples

use std::time::Duration;
use ruststream_fred::RedisStream;

// Fresh tail: a normal worker reading new entries.
let fresh = RedisStream::new("orders").group("workers").count(128);

// Recovery: reclaim entries a crashed worker left pending for over 30s.
let recover = RedisStream::reclaim("orders", Duration::from_secs(30)).group("workers");

Implementations

impl RedisStream

pub fn new(key: impl Into<String>) -> Self

A fresh-tail subscription on key: reads new entries via XREADGROUP >.

A consumer group is required; set it with group.

pub fn reclaim(key: impl Into<String>, min_idle: Duration) -> Self

A recovery subscription on key: reclaims pending entries idle at least min_idle via XAUTOCLAIM. Run it alongside a new subscriber on the same group to pick up messages a consumer fetched but died before acking.

min_idle has no default and must exceed the longest legitimate handler runtime: set it too low and a healthy consumer’s in-flight message gets reclaimed and processed twice.

pub fn group(self, group: impl Into<String>) -> Self

Sets the consumer group. Required for every subscription.

pub fn consumer(self, consumer: impl Into<String>) -> Self

Sets this consumer’s name within the group. Defaults to an auto-generated unique name.

pub const fn count(self, count: u64) -> Self

Upper bound on entries fetched per read. Defaults to 64.

pub const fn block(self, block: Duration) -> Self

How long one read blocks waiting for entries. Defaults to 5 seconds. In fresh-tail mode this is the XREADGROUP server-side block; in reclaim mode XAUTOCLAIM does not block, so this is the poll interval slept between scans that find nothing to reclaim.

pub fn start_id(self, start: StreamStart) -> Self

Where a newly created group starts reading. Ignored if the group already exists. Only meaningful for the fresh-tail new mode.

pub fn dead_letter(self, key: impl Into<String>) -> Self

Routes dropped and poison messages to the named dead-letter stream instead of discarding them. Off by default. The copy is tagged with DEAD_LETTER_REASON_HEADER. See [crate::deadletter].

pub const fn max_deliveries(self, max: u64) -> Self

Caps how many times a message may be delivered before it is treated as poison (dead-lettered or, with no dead-letter stream, discarded). Off by default.

The cap is checked against both the framework retry-count header (the nack/republish loop) and the native stream delivery count (the reclaim loop), so a message poisoning either way is caught.

pub fn delayed_retry(self, retry: DelayedRetry) -> Self

Opts this subscription into durable, crash-safe delayed retry backed by a ZSET delay queue.

Off by default: without it, retry_after(delay) / nack_after(delay) degrade to the runtime’s broker-agnostic deferred re-publish (at-most-once over the delay window). With it, a delayed delivery is ZADDed to the named ZSET and replayed from there once due, so the retry survives a process crash. See DelayedRetry for the key and TTL requirements.

The sweeper that replays due entries runs inside this subscription’s read loop, so its granularity is the read block interval.

pub fn key(&self) -> &str

The stream key this subscription reads.

Trait Implementations

impl Clone for RedisStream

fn clone(&self) -> RedisStream

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for RedisStream

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl SubscriptionSource<RedisBroker> for RedisStream

type Subscriber = RedisSubscriber

The subscriber type this source opens.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe( self, broker: &RedisBroker, ) -> Result<Self::Subscriber, RedisError>

Opens the subscription against broker. Called once, after Broker::connect.