RedisGroupPosition

RedisGroupPosition names where a consumer group resumes: the oldest retained entry, the tail, or the entry after an id.

Constructor	Where the group resumes
`RedisGroupPosition::beginning()`	the oldest entry the stream still retains
`RedisGroupPosition::end()`	the tail: only entries added afterwards
`RedisGroupPosition::after(id)`	the entry after that `EntryId`, which is not delivered again

A move is XGROUP SETID, which rewrites the group's cursor. Redis keeps one cursor per group, so a move applies to every consumer of that group, not only to the subscription that asked. A subscription moves its group at startup with start_at, and a handler moves it while the service runs with a seek. A delivery's own position is a RedisGroupPosition too, and moving there delivers that entry again and the entries after it. Streams are the form with a cursor: the core's Seekable capability is implemented for a stream subscription. Where a group starts the first time it is created is its StreamStart.

Example

A producer marks the orders queued behind a marker as void, and the worker moves its group to the tail when it reads the marker.

Macros
Manual

examples/redis_positions_position.rs
/// On the void marker the group moves to the tail: the orders queued behind the marker are never
/// processed, by this consumer or any other of `workers`.
#[subscriber(RedisStream::new("orders").group("workers"))]
async fn work(order: &Order, Ctx(seeker): Ctx<keys::SeekHandle>) -> HandlerOutcome {
    if order.id != 0 {
        tracing::info!(order.id, "order processed");
        return HandlerOutcome::ack();
    }
    match seeker.seek(RedisGroupPosition::end()).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

examples/manual/redis_positions_position.rs
/// On the void marker the group moves to the tail: the orders queued behind the marker are never
/// processed, by this consumer or any other of `workers`.
struct Work;

impl Handle<Order, (), (), StreamContext> for Work {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, StreamContext>,
    ) -> Result<(), HandlerOutcome> {
        if order.id != 0 {
            tracing::info!(order.id, "order processed");
            return Ok(());
        }
        ctx.context(keys::SeekHandle)
            .seek(RedisGroupPosition::end())
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let orders = RedisStream::new("orders").group("workers");
        b.include(subscriber(orders, Work).build());
    })
}

Example​

Example