start_at

start_at moves the group to a position before the first delivery, on every startup; the whole group moves, not one consumer.

It is a clause of the attribute, start_at(RedisGroupPosition::beginning()), or .start_at(position) after build() in the manual form. The move happens whatever the group read and acknowledged before. With beginning() the group reads every retained entry again, the acknowledged ones included. Every consumer of the group follows the move, and a consumer blocked in a read sees it within one block. It suits a read model rebuilt from the retained stream on each boot. A StreamStart is different: it applies only when the subscription creates the group. The block step reads the same before and after start_at.

Example

The audit projection is rebuilt from the oldest retained entry on every startup, although its group acknowledged everything it read before.

Macros

examples/redis_positions_start_at.rs

/// Rebuilds the audit projection from the oldest retained entry on every start. The group's
/// cursor moves, so every consumer of `projection` replays.
#[subscriber(
    RedisStream::new("audit.trail").group("projection"),
    start_at(RedisGroupPosition::beginning())
)]
async fn project(entry: &AuditEntry) -> HandlerOutcome {
    tracing::info!(entry.id, "entry projected");
    HandlerOutcome::ack()
}

Manual

examples/manual/redis_positions_start_at.rs

/// Projects every audit entry it reads.
struct Project;

impl Handle<AuditEntry> for Project {
    fn handle(
        &self,
        entry: &AuditEntry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(entry.id, "entry projected");
        ready(Ok(()))
    }
}

fn app(url: &str) -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(RedisBroker::standalone(url), |b| {
        // The projection is rebuilt from the oldest retained entry on every start.
        let trail = RedisStream::new("audit.trail").group("projection");
        b.include(
            subscriber(trail, Project)
                .build()
                .start_at(RedisGroupPosition::beginning()),
        );
    })
}