Skip to main content

processing

processing names the list an entry waits on while handled; on a cluster both keys share one hash tag, or the subscription refuses to start.

The processing list is where a reliable subscription moves an entry before its handler runs. Its default name is the queue's key with .processing appended, so jobs waits on jobs.processing. Every consumer of the queue that uses the same name shares that list. On a cluster a claim moves the entry between the two keys in one command, and Redis refuses a command that spans hash slots. A hash tag puts both keys on one slot: Redis hashes only the part between { and }, so {exports} and {exports}.in-flight land together. The default name follows the tag of its queue. A subscription whose two keys hash to different slots does not start, and its error names the tag to use.

Example​

The export queue runs on a cluster, and its processing list carries the queue's tag.

examples/redis_lists_processing.rs
/// `{exports}` is the hash tag of both keys, so the queue and its processing list live on one
/// slot of the cluster.
#[subscriber(
    RedisList::new("{exports}")
        .reliable()
        .processing("{exports}.in-flight")
)]
async fn export(job: &Export) -> HandlerOutcome {
    tracing::info!(job.id, "export written");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("exports", "0.1.0")).with_broker(
        RedisBroker::cluster(["redis://localhost:7000"]),
        |b| {
            b.include(export);
        },
    )
}