Struct RedisList 

pub struct RedisList { /* private fields */ }

Describes one list subscription against a ConnectedRedisBroker.

Examples

use std::time::Duration;

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

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

// Reliable: the entry waits on a processing list until the handler acknowledges it.
#[subscriber(RedisList::new("jobs").reliable().block(Duration::from_secs(2)))]
async fn run_job(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "running");
    HandlerOutcome::ack()
}

// Simple: popped and gone, so a crash mid-handler loses the entry.
#[subscriber(RedisList::new("metrics"))]
async fn record(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "recorded");
    HandlerOutcome::ack()
}

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

Implementations

impl RedisList

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

A simple (at-most-once) BRPOP work-queue consumer on key.

pub const fn reliable(self) -> Self

Switches to reliable (at-least-once) mode: entries move to a processing list and are removed on ack.

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

Sets the processing-list key used in reliable mode. Defaults to <key>.processing.

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

How long one blocking pop waits before looping. Defaults to 5 seconds.

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

pub fn codec(self, codec: impl Codec + 'static) -> Self

Decodes the header/payload envelope with codec (must match the publisher). Without it the default binary framing is used. Either way the payload arrives as it was published.

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

How long a claimed reliable-mode entry may sit idle on the processing list before the recovery watchdog returns it to the main list. Required for (and only meaningful with) recovery_zset.

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

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

Opts reliable mode into orphan recovery, naming the ZSET key that tracks in-flight claims.

Off by default (a dead consumer’s entry stays stranded on the processing list). The key has no sane default, so it is named explicitly here; pair it with min_idle, which is required when recovery is on. Reliable mode is implied. The Recovery page shows the watchdog in a service.

Examples

use std::time::Duration;

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

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

// At-least-once, with the watchdog that returns a dead consumer's entry to the queue. Set
// `min_idle` above the longest handler runtime, or a running job is recovered and repeated.
#[subscriber(
    RedisList::new("jobs")
        .reliable()
        .min_idle(Duration::from_secs(30))
        .recovery_zset("jobs.inflight")
)]
async fn run_job(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "running");
    HandlerOutcome::ack()
}

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

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

An optional auto-cleanup TTL on the recovery ZSET key (refreshed on every claim).

When set it must exceed min_idle (and the longest legitimate handler runtime), or in-flight tracking is dropped before the watchdog can act.

pub fn key(&self) -> &str

The list key this subscription consumes.

Trait Implementations

impl Clone for RedisList

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 RedisList

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

Formats the value using the given formatter.

impl IntoSource for RedisList

type Source = RedisList

The source the constructor stores.

fn into_source(self) -> Self

Builds the source.

impl RedeliveryAddressed<ConnectedRedisBroker> for RedisList

The list key: the subscription pops from it and RedisListPublish pushes onto it, so a deferred copy lands in the same queue the delivery came from.

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

Where a publish reaches this subscription again.

impl SubscriptionSource<ConnectedRedisBroker> for RedisList

type Copies = AddressedCopies

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

type Subscriber = RedisListSubscriber

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.