RedisListPublish

RedisListPublish, Publish in the list prelude, pushes with LPUSH and can re-arm a key expiry at every publish.

The message's name is the list key. A list consumer pops from the other end, so the list is a first-in, first-out queue. Headers travel beside the payload in a binary frame by default; codec frames them with a codec the consumer also sets.

ttl sets a time to live on the list key and arms it again with PEXPIRE at every publish. A queue in use never expires; a queue nobody pushes to for the whole interval does, with whatever it still holds. The expiry is the key's: Redis lists expire as a whole, never per element. The push and the PEXPIRE leave in one round trip but are two commands, not one transaction.

The policy pairs into a RedisListPublisher at startup.

Example

Each order asks for a shipping label on the label-jobs list, whose key expires after an hour without a push.

Macros
Manual

examples/redis_publishing_list.rs
#[subscriber(RedisStream::new("orders").group("labels"), reply)]
async fn request_label(order: &Order) -> LabelJob {
    LabelJob { order_id: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    // A queue nobody drains for an hour expires instead of growing.
    let jobs = list::Publish::new().ttl(Duration::from_secs(3600));
    RustStream::new(AppInfo::new("labels", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(request_label).out_reply(jobs);
        },
    )
}

examples/manual/redis_publishing_list.rs
struct RequestLabel;

impl Handle<Order, LabelJob> for RequestLabel {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<LabelJob, HandlerOutcome>> {
        ready(Ok(LabelJob { order_id: order.id }))
    }
}

fn app() -> impl App {
    // A queue nobody drains for an hour expires instead of growing.
    let jobs = list::Publish::new().ttl(Duration::from_secs(3600));
    RustStream::new(AppInfo::new("labels", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            let request = subscriber(RedisStream::new("orders").group("labels"), RequestLabel);
            b.include(request.reply().build()).out_reply(jobs);
        },
    )
}

Example​

Example