StreamStart

StreamStart says where a consumer group starts when the subscription creates it: the beginning, the end, or after an id.

A descriptor names it with start_id. StreamStart::New is the default: the group reads only entries added after it was created. StreamStart::Beginning makes the group read every entry the stream still holds, then the new ones. StreamStart::Id(id) starts the group after that entry id, which it does not deliver. The setting is read once, when the group is created. A group that already exists keeps its own cursor, whatever a later start names. A group that has to move on every start is a start_at subscription.

Example

A reporting group added to a running system reads the orders the stream already holds, the first time it is created.

Macros
Manual

examples/redis_streams_start.rs
/// A reporting group added to a running system: the first time it is created, it reads every
/// order the stream still holds, then the new ones.
#[subscriber(RedisStream::new("orders").group("reports").start_id(StreamStart::Beginning))]
async fn report(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order reported");
    HandlerOutcome::ack()
}

examples/manual/redis_streams_start.rs
/// Reports every order it reads.
struct Report;

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

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(broker, |b| {
        // The first time the group is created, it reads every order the stream still holds.
        let orders = RedisStream::new("orders")
            .group("reports")
            .start_id(StreamStart::Beginning);
        b.include(subscriber(orders, Report).build());
    })
}

Example​

Example