block

block sets how long one read waits on the server: the XREADGROUP BLOCK on a stream, the BRPOP or BLMOVE timeout on a list.

The wait is five seconds unless a subscription names another. A descriptor names it with RedisStream::block or RedisList::block. The mount site names it with the block step, which overrides the descriptor. The step chains after another step of the mount chain: after .batch(n), after .pipeline(), or, in the manual form, after build(). It reads the same before and after start_at(..). On a reclaim subscription the wait is the pause between two XAUTOCLAIM scans that found nothing. A read wakes as soon as an entry arrives, so the wait costs nothing on a busy stream. It decides how soon an idle read sees what happens beside it: a seek of its group, or a due entry of a delay queue, which the subscription sweeps once per read. A shorter wait sees them sooner, at the price of one more round trip per idle wait. The step exists on stream and list subscriptions, the two forms whose read blocks.

Example

Orders wait at most a second per read; invoices are billed sixteen at a time, with a two-second wait set at the mount site.

Macros
Manual

examples/redis_subscriptions_block.rs
/// Each read of `orders` waits at most one second on the server.
#[subscriber(RedisStream::new("orders").group("workers").block(Duration::from_secs(1)))]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

/// Bills up to 16 orders a read.
#[subscriber(RedisStream::new("invoices").group("billing"))]
async fn bill(orders: &[Order]) -> HandlerOutcome {
    tracing::info!(count = orders.len(), "orders billed");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(place);
        // The size is the framework's step and the wait is Redis's, chained after it.
        b.include(bill.batch(nonzero!(16)).block(Duration::from_secs(2)));
    })
}

examples/manual/redis_subscriptions_block.rs
/// Places one order.
struct Place;

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

/// Bills the orders of one read.
struct Bill;

impl Handle<[Order]> for Bill {
    fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = orders.len(), "orders billed");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // Each read of `orders` waits at most one second on the server.
        let orders = RedisStream::new("orders")
            .group("workers")
            .block(Duration::from_secs(1));
        b.include(subscriber(orders, Place).build());
        let invoices = RedisStream::new("invoices").group("billing");
        b.include(
            subscriber(invoices, Bill)
                .batch(nonzero!(16))
                .build()
                .block(Duration::from_secs(2)),
        );
    })
}

Example​

Example