Clustered actors

Finding the device manager

Find the device manager by its key alone, wherever it runs, through the receptionist.

The first node of the previous step looked the device manager up at the address of its node. In a real cluster, the node that asks rarely knows which node runs what. In this step the first node finds the device manager by its key alone.

Subscribing to a key

tellus/examples/tutorials/clustered/03_receptionist.rs

async fn find_manager() -> anyhow::Result<ActorRef<Command>> {
    let mut managers =
        receptionist::subscribe(&Key::new(MANAGER_KEY)).context("subscribing to the key")?;
    let find = async {
        loop {
            let current = managers
                .current()
                .context("resolving the device managers")?;
            match current.into_iter().next() {
                Some(manager) => return anyhow::Ok(manager),
                None => managers.changed().await,
            }
        }
    };

    timeout(FIND_TIMEOUT, find)
        .await
        .context("no device manager registered within the timeout")?
}

receptionist::subscribe returns a Subscription to every actor registered under the key, on any member. current resolves the set as of now into references, and changed waits until it has changed since the last current. So instead of polling, find_manager waits for the next registration, unregistration or change of membership.

The set only holds actors on members that are up. A registration disappears when its actor terminates or its node is gone, so a subscription also tells you when an actor you found is no longer there.

The first node

The first node now calls find_manager() instead of lookup_manager(seeds[1]); everything else is unchanged:

tellus/examples/tutorials/clustered/03_receptionist.rs

async fn first_node() -> anyhow::Result<()> {
    let seeds = reserved_addrs(2)?;
    let _manager_node = spawn_node(MANAGER_NODE_ARG, seeds[1], &seeds)?;
    join_cluster(seeds[0], seeds.clone()).await?;
    let manager = find_manager().await?;

    let readings = [
        ("attic", "left", 21.5),
        ("attic", "right", 19.1),
        ("cellar", "boiler", 11.0),
        ("cellar", "boiler", f64::NAN),
    ];
    for (group_name, device_name, degrees) in readings {
        manager.tell(Command::Record {
            group_name: group_name.to_string(),
            device_name: device_name.to_string(),
            degrees,
        });
    }
    manager.tell(Command::Retire {
        group_name: "attic".to_string(),
        device_name: "right".to_string(),
    });

    for group_name in ["attic", "cellar", "garage"] {
        let average = manager
            .ask(ASK_TIMEOUT, |reply_to| Command::Average {
                group_name: group_name.to_string(),
                reply_to,
            })
            .await
            .context("asking for the average temperature")?;
        match average {
            Some(degrees) => println!("Average temperature in {group_name}: {degrees:.2}"),
            None => println!("No readings in {group_name}"),
        }
    }
    Ok(())
}

Run it

From a clone of the tellus repository:

cargo run --quiet -p tellus --features cluster-dev --example clustered_03_receptionist

It prints:

Average temperature in attic: 21.50
No readings in cellar
No readings in garage

The output is the same as in the previous step, but nothing in the first node depends on where the device manager runs anymore.

What you learned

The receptionist finds actors by key on any member. A subscription follows the set of registered actors as registrations, terminations and members come and go. Next, there are two device managers, and one of them goes away for good.