Commands

RedisPipeline queues fred's commands, Lua through eval and evalsha, and any other command through custom; a reply is not read back.

The pipeline carries fred's command methods for keys, strings, hashes, lists, sets, sorted sets, streams, functions and PUBLISH, under fred's names and arguments. Each method queues its command into the delivery's segment and returns Result<(), fred::error::Error>. Ok means queued; an error is fred's refusal at queue time, such as an argument that does not convert. The command runs after the handler returns, so its reply never reaches the handler. A command whose answer the handler needs runs on the pool instead, through FredPool.

eval queues a Lua script by its source and evalsha by its SHA1 digest, each with its keys and arguments. On a cluster a script goes to the node of its first key. custom queues any command by name, for one the methods do not cover. The first command a delivery queues opens a fred pipeline for it; a delivery that queues nothing allocates none. The in-process Redis of the test harness runs no scripts, so a test of a script runs against a server.

Example

The handler queues two commands: a push onto the shipping queue and a counter.

Macros
Manual

examples/redis_pipelining_commands.rs
/// Puts the order on the shipping queue and counts it, both in the delivery's round.
#[subscriber(RedisStream::new("orders").group("workers"))]
async fn ship(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
    // `Ok` means queued: the commands run after this handler acknowledges the delivery.
    if pipeline
        .lpush("shipments", order.id.to_string())
        .await
        .is_err()
        || pipeline
            .hincrby("orders:totals", "shipped", 1)
            .await
            .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/redis_pipelining_commands.rs
/// Puts the order on the shipping queue and counts it, both in the delivery's round.
struct Ship;

impl Handle<Order, (), (), PipelineContext> for Ship {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, PipelineContext>,
    ) -> Result<(), HandlerOutcome> {
        let pipeline = ctx.context(keys::Pipeline);
        // `Ok` means queued: the commands run after this handler acknowledges the delivery.
        if pipeline
            .lpush("shipments", order.id.to_string())
            .await
            .is_err()
            || pipeline
                .hincrby("orders:totals", "shipped", 1)
                .await
                .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

The same two writes as one Lua script, so that no client sees one without the other:

Macros
Manual

examples/redis_pipelining_commands.rs
/// The same two writes as one Lua script, so no client sees one without the other.
const SHIP_AND_COUNT: &str = "redis.call('HINCRBY', KEYS[2], 'shipped', 1) \
                              return redis.call('LPUSH', KEYS[1], ARGV[1])";

#[subscriber(RedisStream::new("orders.priority").group("workers"))]
async fn ship_first(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
    let touched = vec!["shipments", "orders:totals"];
    if pipeline
        .eval(SHIP_AND_COUNT, touched, vec![order.id.to_string()])
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/redis_pipelining_commands.rs
/// The same two writes as one Lua script, so no client sees one without the other.
const SHIP_AND_COUNT: &str = "redis.call('HINCRBY', KEYS[2], 'shipped', 1) \
                              return redis.call('LPUSH', KEYS[1], ARGV[1])";

struct ShipFirst;

impl Handle<Order, (), (), PipelineContext> for ShipFirst {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, PipelineContext>,
    ) -> Result<(), HandlerOutcome> {
        let touched = vec!["shipments", "orders:totals"];
        let pipeline = ctx.context(keys::Pipeline);
        if pipeline
            .eval(SHIP_AND_COUNT, touched, vec![order.id.to_string()])
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Example​

Example