Struct RedisGroupSeeker 

pub struct RedisGroupSeeker { /* private fields */ }

Moves a consumer group’s cursor.

The handle behind the core Seekable capability, minted with Seekable::seeker or read out of a delivery’s context under the SeekHandle key - as a Ctx(seeker): Ctx<keys::SeekHandle> handler parameter, or with ctx.context(keys::SeekHandle).

A seek is group-wide

Redis keeps one cursor per consumer group, so XGROUP SETID moves the read position for every consumer of that group, not only for the subscription this seeker came from. That is unlike a partitioned log, where a seek is scoped to one consumer instance. Treat a seek as an operation on the group: replaying a range replays it for the whole worker pool, and skipping forward skips for all of them.

What a seek does not do

• It does not clear the pending entries list. Entries already delivered and not acknowledged stay pending and remain reachable through the reclaim path (RedisStream::reclaim), whichever way the cursor moved.
• It does not cancel delayed retries. Copies already scheduled in a DelayedRetry::DurableZset queue are keyed by their due time, not by the cursor, so they are appended to the stream when they fall due regardless of where the group is reading.
• It does not carry the native delivery count over. A replayed entry is read as a new delivery, and Redis counts it from one again, so a max_attempts declaration at the mount site that reads this count caps from the replay on; the framework retry-count header only moves on an actual retry.

Timing

The cursor changes as soon as seek returns, but a subscription parked in a blocking XREADGROUP observes it only on its next read - within one RedisStream::block interval. Entries selected under the old cursor are discarded rather than delivered, so a seek never yields a message from the position it moved away from.

Examples

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

#[derive(Deserialize)]
struct Event {
    kind: String,
}

// A `rebuild` event makes the projection group read the whole retained stream again.
#[subscriber(RedisStream::new("events").group("projection"))]
async fn project(event: &Event, Ctx(seeker): Ctx<keys::SeekHandle>) -> HandlerOutcome {
    if event.kind != "rebuild" {
        tracing::info!(event.kind, "projecting");
        return HandlerOutcome::ack();
    }
    match seeker.seek(RedisGroupPosition::beginning()).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

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

Implementations

impl RedisGroupSeeker

pub fn key(&self) -> &str

The stream key this handle repositions a group over.

pub fn group(&self) -> &str

The consumer group whose cursor this handle moves.

The cursor belongs to the group, so this names everything a seek through this handle repositions.

Trait Implementations

impl Clone for RedisGroupSeeker

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 RedisGroupSeeker

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

Formats the value using the given formatter.

impl Seeker for RedisGroupSeeker

async fn seek(&self, to: RedisGroupPosition) -> Result<(), RedisError>

Moves the group cursor with XGROUP SETID.

Errors

Returns RedisError::Stream when the group or the stream does not exist, or the command fails, and RedisError::ShutDown once the broker was shut down.

type Position = RedisGroupPosition

The broker’s own position type (Kafka partition offsets, a Redis entry id, a byte offset), constructed by the broker crate or captured from a delivered message via Positioned::position.

type Error = RedisError

The error returned when the broker rejects the reposition.