Struct RedisStream 

pub struct RedisStream { /* private fields */ }

Describes one Redis Streams subscription against a ConnectedRedisBroker.

Examples

use std::time::Duration;

use ruststream_fred::stream::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

// Fresh tail: the normal worker, reading new entries.
#[subscriber(RedisStream::new("orders").group("workers"))]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order");
    HandlerOutcome::ack()
}

// Recovery on the same group: entries a crashed worker left pending for over 30s.
#[subscriber(RedisStream::reclaim("orders", Duration::from_secs(30)).group("workers"))]
async fn recover(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "recovered");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(handle);
            b.include(recover);
        },
    )
}

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.

Examples

A plain stream has no delivery count of its own, so a retry cap reads the framework’s retry-count header on the copies the runtime publishes:

use ruststream_fred::stream::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

// The handler counts nothing itself.
#[subscriber(RedisStream::new("orders").group("workers"))]
async fn handle(order: &Order) -> HandlerOutcome {
    if order.id == 0 {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            // Five deliveries per message, counting the first. The fifth failure sends it to
            // "orders.dlq" instead of back to "orders".
            b.include(handle)
                .max_attempts(nonzero!(5u32))
                .dead_letter("orders.dlq");
        },
    )
}

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.

Examples

use std::time::Duration;

use ruststream_fred::stream::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[subscriber(RedisStream::new("orders").group("workers"))]
async fn handle(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "processing");
    HandlerOutcome::ack()
}

// Recovery on a server older than 8.4: a reclaim subscription beside the fresh-tail one, on the
// same group, takes the orders a dead worker fetched and never finished.
#[subscriber(RedisStream::reclaim("orders", Duration::from_secs(30)).group("workers"))]
async fn recover(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "recovered");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(handle);
            b.include(recover);
        },
    )
}

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

A subscription on key that claims and reads in one call: XREADGROUP ... CLAIM first takes over the group’s entries idle at least min_idle, longest idle first, then fills the rest of the read with fresh ones. One subscription, one consumer, one handler.

Every delivery carries the server’s own IDLE_MS_HEADER and DELIVERY_COUNT_HEADER, both as of the delivery before this one: a fresh entry reports zero for both, and the first claim back reports one earlier delivery. The count is therefore the number of attempts that did not finish, which is what a handler reads to give up on a message itself.

This is Redis’s own redelivery, so this mode is the one that settles a retry natively. retry() leaves the entry in the pending entries list rather than appending a copy, so the next read brings it back under its own id with the count one higher, no sooner than min_idle from now. retry_after(delay) takes the same path and the entry comes back after min_idle, so a delay shorter than that rounds up to it; name a delayed_retry queue to have the delay honoured exactly instead. The delivery reports its count to the runtime, which is what a max_attempts declaration at the mount site caps.

min_idle has no default and must exceed the longest legitimate handler runtime: below it, a message a healthy consumer is still working on is taken away and processed twice.

Redis 8.4.0 added the option. A registration of this mode against an older server refuses to start, naming both versions; on such a server use new with a reclaim subscription beside it.

Examples

use std::time::Duration;

use ruststream_fred::stream::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Order {
    id: u64,
    in_stock: bool,
}

// New orders and the ones a dead worker took and never finished, in one subscription. A
// retry stays pending and comes back after 30s under its own id.
#[subscriber(RedisStream::claiming("orders", Duration::from_secs(30)).group("workers"))]
async fn handle(order: &Order) -> HandlerOutcome {
    if !order.in_stock {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, "order");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            // The server's delivery count feeds the cap.
            b.include(handle)
                .max_attempts(nonzero!(5u32))
                .dead_letter("orders.dlq");
        },
    )
}

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 block(self, block: Duration) -> Self

How long one read blocks waiting for entries. Defaults to 5 seconds. In the fresh-tail and claiming modes this is the XREADGROUP server-side block, which wakes on a new entry and, under CLAIM, on an entry ageing past the threshold; in reclaim mode XAUTOCLAIM does not block, so this is the poll interval slept between scans that find nothing to reclaim.

Also reachable at the mount site, after the batch size, through RedisSubscribeExt.

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 modes that read the tail, new and claiming.

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, which the mount site binds with out_retry and which is 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) -> Self

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 IntoSource for RedisStream

type Source = RedisStream

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl RedeliveryAddressed<ConnectedRedisBroker> for RedisStream

The stream key, in every read mode: an XADD there is read by the consumer group this subscription belongs to.

Under reclaim the copy is read by the group’s tail reader rather than by this subscription, which only ever takes entries already pending on another consumer. That is the topology the mode is written for - a reclaim subscriber runs beside a new one on the same group - so the copy reaches the handler either way.

fn redelivery_address( &self, _connected: &ConnectedRedisBroker, ) -> impl Future<Output = Result<RedeliveryAddress, RedisError>>

Where a publish reaches this subscription again.

impl SubscriptionSource<ConnectedRedisBroker> for RedisStream

type Copies = AddressedCopies

A retry copy is an XADD this process makes, and the stream key is where it goes.

type Subscriber = RedisSubscriber

The subscriber type this source opens.

fn name(&self) -> &str

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

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

Takes the registration’s retry declaration into the descriptor, before the subscription opens.

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

Opens the subscription against the connected broker. Called once at startup.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.