Flush

A window flushes when the read's COUNT has settled, when nothing is outstanding, or when the subscription stops.

The flush sends the segments of the acknowledged deliveries, then one pipeline of settles, on a connection of the pool apart from the one the reads block on. A stream's settles leave as one XACK, a reliable list's as one LREM per entry. Under load that is one round trip per fetched batch; on a trickle the window leaves as soon as the last handler of the read returns. What a handler queued follows its delivery's outcome:

Outcome	The handler's commands	The settle
`ack`	sent, before the settle	the form's own: `XACK`, `LREM`
`drop`	dropped	sent alone
`retry`, `retry_after`	dropped	the retry's: the copy or the `ZADD`, then `XACK`
unsettled	dropped	none: the entry stays pending

Pub/Sub and a simple list settle nothing, so on them every outcome but ack drops the segment. A flush that fails is reported once on the delivery stream, with the subscription, the number of commands and settles it carried, and how many of them failed. On the forms that settle, an entry whose settle failed stays pending and is delivered again. The Window page draws the flush and the Settlement page what each outcome becomes per form.

A batch mount's segment is the whole batch. The body's commands and the settles of every entry leave together, and the commands only if every entry is acknowledged. A pipelined reliable list reads its batch with one pipeline of LMOVE, a simple one with RPOP key count. On an empty queue both wait with their blocking pop, BLMOVE or BRPOP, for the next entry.

Example

A batch handler records every order of the batch; one order without items drops the batch, and every push it queued.

Macros
Manual

examples/redis_pipelining_flush.rs
/// Records every order of a batch; one order without items drops the batch and its pushes.
#[subscriber(RedisStream::new("orders").group("workers"))]
async fn accept_all(orders: &[Order], ctx: &mut Context<'_, PipelineContext>) -> HandlerOutcome {
    let pipeline = ctx.context(keys::Pipeline);
    for order in orders {
        if pipeline
            .lpush("orders:accepted", order.id.to_string())
            .await
            .is_err()
        {
            return HandlerOutcome::retry();
        }
    }
    if orders.iter().any(|order| order.items == 0) {
        tracing::warn!(
            orders = orders.len(),
            "a batch with an empty order is dropped"
        );
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(accept_all.batch(nonzero!(64)).pipeline());
        },
    )
}

examples/manual/redis_pipelining_flush.rs
/// Records every order of a batch; one order without items drops the batch and its pushes.
struct AcceptAll;

impl Handle<[Order], (), (), PipelineContext> for AcceptAll {
    async fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        ctx: &mut Context<'_, PipelineContext>,
    ) -> Result<(), Vec<HandlerOutcome>> {
        let pipeline = ctx.context(keys::Pipeline);
        for order in orders {
            if pipeline
                .lpush("orders:accepted", order.id.to_string())
                .await
                .is_err()
            {
                return Err(orders.iter().map(|_| HandlerOutcome::retry()).collect());
            }
        }
        if orders.iter().any(|order| order.items == 0) {
            tracing::warn!(
                orders = orders.len(),
                "a batch with an empty order is dropped"
            );
            return Err(orders.iter().map(|_| HandlerOutcome::drop()).collect());
        }
        Ok(())
    }
}

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

Example​

Example