bind

pipeline.bind(out) makes a slot's publishes join the delivery's round, so they leave only if the delivery is acknowledged.

The handler takes the slot as Out<impl Bindable>, where Bindable is the bound every publisher of this crate meets. bind hands the slot back with its codec and transforms, and every publish through it from then on is queued into the round. A bound publish leaves with the delivery's settle, after the commands the handler queued before it. A slot publish the handler does not bind leaves at once, whatever the delivery's outcome. The slot's policy is one of this crate's: stream::Publish, list::Publish or pubsub::Publish; another broker's publisher does not compile there. A reply joins the round through the InRound transform.

Example

The handler writes an order's first step to the order log through a bound slot, so a dropped order leaves no entry.

Macros

examples/redis_pipelining_round.rs

#[subscriber(RedisStream::new("orders").group("workers"), reply)]
async fn accept(
    order: &Order,
    Ctx(pipeline): Ctx<keys::Pipeline>,
    Out(out): Out<impl Bindable>,
) -> Result<LogEntry, HandlerOutcome> {
    // Bound to the round, the entry leaves with the acknowledgement, and not without it.
    let log = pipeline.bind(out);
    let accepted = LogEntry {
        id: order.id,
        step: Step::Accepted,
    };
    if log.message(&accepted).publish().await.is_err() {
        return Err(HandlerOutcome::retry());
    }
    if order.items == 0 {
        tracing::warn!(order.id, "an order without items is dropped");
        return Err(HandlerOutcome::drop());
    }
    Ok(LogEntry {
        id: order.id,
        step: Step::Confirmed,
    })
}

Manual

examples/manual/redis_pipelining_round.rs

struct Accept;

impl<Log> Handle<Order, LogEntry, Outs<(Log,)>, PipelineContext> for Accept
where
    Log: OutEntry<DefaultSlot, Wire: Bindable>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Log,)>,
        ctx: &mut Context<'_, PipelineContext>,
    ) -> Result<LogEntry, HandlerOutcome> {
        // Bound to the round, the entry leaves with the acknowledgement, and not without it.
        let log = ctx.context(keys::Pipeline).bind(outs.get(DefaultSlot));
        let accepted = LogEntry {
            id: order.id,
            step: Step::Accepted,
        };
        if log.message(&accepted).publish().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        if order.items == 0 {
            tracing::warn!(order.id, "an order without items is dropped");
            return Err(HandlerOutcome::drop());
        }
        Ok(LogEntry {
            id: order.id,
            step: Step::Confirmed,
        })
    }
}