pipeline

.pipeline() settles a subscription in a window, and a handler queues its own Redis commands to leave with its acknowledgement.

The step chains where the handler is mounted, b.include(handler.pipeline()), like .workers(..) or .block(..). It applies to a RedisStream, a RedisList, a RedisPubSub and a RedisPubSubPattern, and keeps every setting the descriptor carries. Each form's prelude brings the step in with RedisPipelineSteps, together with the InRound transform and the Bindable bound.

A handler takes the delivery's round with Ctx<keys::Pipeline>, a RedisPipeline. What it queues there runs after the handler returns, in one pipeline with the settles of the whole window: under load, one round trip per fetched batch instead of one per command. The queued commands follow the delivery's outcome: they leave when the delivery is acknowledged and are dropped otherwise. So a handler's Redis side effects never land for a delivery that comes back. The Window page draws a window from the read to the flush.

A handler that reads Ctx<keys::Pipeline> and is mounted without .pipeline() does not compile. The error is the framework's general one: the handler's context, PipelineContext, does not match the subscription's. A delivery whose handler queues nothing costs the window no buffer. .atomic() after the step makes each delivery's segment a transaction.

Example

An accepted order is recorded on a list in the same round trip as its XACK; an order without items is dropped, and its push with it.

Macros
Manual

examples/redis_pipelining.rs
/// Records an accepted order on the `orders:accepted` list, in the same round trip as its `XACK`.
#[subscriber(RedisStream::new("orders").group("workers"))]
async fn accept(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
    if pipeline
        .lpush("orders:accepted", order.id.to_string())
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    if order.items == 0 {
        // The delivery is dropped, and the push queued above goes with it.
        tracing::warn!(order.id, "an order without items 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.pipeline());
        },
    )
}

examples/manual/redis_pipelining.rs
/// Records an accepted order on the `orders:accepted` list, in the same round trip as its `XACK`.
struct Accept;

impl Handle<Order, (), (), PipelineContext> for Accept {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, PipelineContext>,
    ) -> Result<(), HandlerOutcome> {
        let pipeline = ctx.context(keys::Pipeline);
        if pipeline
            .lpush("orders:accepted", order.id.to_string())
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        if order.items == 0 {
            // The delivery is dropped, and the push queued above goes with it.
            tracing::warn!(order.id, "an order without items is dropped");
            return Err(HandlerOutcome::drop());
        }
        Ok(())
    }
}

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

Example​

Example