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 injected into a handler as a Seek(seeker): Seek<RedisGroupSeeker> parameter.

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 reset delivery counts. A replayed entry is delivered again, so its native delivery count (the one the reclaim path reads and RedisStream::max_deliveries caps) grows with each replay; the framework retry-count header only moves on an actual nack.

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::{Broker, Seekable, Seeker};
use ruststream_fred::{RedisBroker, RedisGroupPosition, RedisStream};

let connected = RedisBroker::standalone("redis://localhost:6379").connect().await?;
let subscriber = connected
    .subscribe(RedisStream::new("orders").group("workers"))
    .await?;

// Minted before the stream opens; usable while it runs.
let seeker = subscriber.seeker();
seeker.seek(RedisGroupPosition::beginning()).await?;

Trait Implementations

impl Clone for RedisGroupSeeker

fn clone(&self) -> RedisGroupSeeker

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.

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.