Cluster

RedisBroker::cluster takes seed nodes, of which one reachable node is enough; the client discovers the rest of the cluster.

Each seed is a host:port address, and a seed without a port means 6379. At startup the client reads the slot map from the first seed that answers. From then on it sends each command to the node that owns its key's hash slot. A cluster seed list carries no password, so a cluster that requires one takes credentials.

Redis refuses a command or a transaction whose keys live in different hash slots. So the transactional policy TransactionalPublish refuses a cluster, and a service that names it does not start. A reliable list moves an entry between its queue and its processing list in one command, so both keys need one slot. A hash tag gives them one: in {jobs} and {jobs}.processing only jobs is hashed. A reliable list whose two keys hash apart refuses to start, and the error names the hash tag that fixes it. In #[outgoing(name = "..")] braces mark a name template, so a type published to a tagged key declares no name, and the call site names the key.

A TestApp test runs a cluster service in process. The in-process Redis keeps the cluster topology and refuses what spans hash slots, as a cluster does.

Example

The service works a reliable list under the hash tag {jobs} and reaches the cluster through two seeds.

Macros
Manual

examples/redis_broker_cluster.rs
/// The hash tag `{jobs}` keeps the queue and its processing list `{jobs}.processing` on one node.
#[subscriber(RedisList::new("{jobs}").reliable())]
async fn run_job(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "job done");
    HandlerOutcome::ack()
}

fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(broker, |b| {
        b.include(run_job);
    })
}

/// Two seeds: the client reaches the cluster through whichever answers first.
#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(RedisBroker::cluster([
        "redis-1.internal:6379",
        "redis-2.internal:6379",
    ]))
}

examples/manual/redis_broker_cluster.rs
struct RunJob;

impl Handle<Job> for RunJob {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(job.id, "job done");
        ready(Ok(()))
    }
}

/// The hash tag `{jobs}` keeps the queue and its processing list `{jobs}.processing` on one node.
fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber(RedisList::new("{jobs}").reliable(), RunJob).build());
    })
}

/// Two seeds: the client reaches the cluster through whichever answers first.
fn app() -> impl App<State = ()> {
    service(RedisBroker::cluster([
        "redis-1.internal:6379",
        "redis-2.internal:6379",
    ]))
}

Example​

Example