Getting started

Watching devices

Learn when an actor has terminated, and rely on having seen all of its messages.

Devices get replaced. When a device retires, it hands over its last reading to the group and stops. The group should then forget the device. In this step the group watches its devices, so tellus tells it when one has terminated.

Retiring a device

A device now also understands Retire. It replies with its last reading and stops:

tellus/examples/tutorials/getting_started/04_watch.rs

fn receive(
    &self,
    _: &ActorContext<Self::Message>,
    incoming: Incoming<Self::Message>,
    last_reading: Self::State,
) -> Result<Control<Self::State>, Self::Error> {
    let Incoming::Message(message) = incoming else {
        unreachable!("the device watches no actor, hence never gets a terminated signal")
    };

    match message {
        DeviceMessage::Record(degrees) => Ok(Control::Continue(Some(degrees))),

        DeviceMessage::Read(reply_to) => {
            reply_to.reply(last_reading);
            Ok(Control::Continue(last_reading))
        }

        DeviceMessage::Retire(reply_to) => {
            reply_to.reply(last_reading);
            Ok(Control::Stop)
        }
    }
}

The group gets two new messages: Retire from the outside, and Retired, which carries a device's handover:

tellus/examples/tutorials/getting_started/04_watch.rs

enum GroupMessage {
    Record {
        device_name: &'static str,
        degrees: f64,
    },

    Read {
        device_name: &'static str,
        reply_to: ReplyTo<Option<f64>>,
    },

    Retire {
        device_name: &'static str,
    },

    Retired {
        device_name: &'static str,
        last_reading: Option<f64>,
    },
}

Watching

tellus/examples/tutorials/getting_started/04_watch.rs

fn receive(
    &self,
    context: &ActorContext<Self::Message>,
    incoming: Incoming<Self::Message>,
    mut devices: Self::State,
) -> Result<Control<Self::State>, Self::Error> {
    match incoming {
        Incoming::Message(GroupMessage::Record {
            device_name,
            degrees,
        }) => {
            let device = devices.entry(device_name).or_insert_with(|| {
                let device = context.spawn(Device);
                context.watch(&device);
                device
            });
            device.tell(DeviceMessage::Record(degrees));
            Ok(Control::Continue(devices))
        }

        Incoming::Message(GroupMessage::Read {
            device_name,
            reply_to,
        }) => {
            match devices.get(device_name) {
                Some(device) => device.tell(DeviceMessage::Read(reply_to)),
                None => reply_to.reply(None),
            }
            Ok(Control::Continue(devices))
        }

        Incoming::Message(GroupMessage::Retire { device_name }) => {
            if let Some(device) = devices.get(device_name) {
                device.tell(DeviceMessage::Retire(context.reply_to(
                    move |last_reading| GroupMessage::Retired {
                        device_name,
                        last_reading,
                    },
                )));
            }
            Ok(Control::Continue(devices))
        }

        Incoming::Message(GroupMessage::Retired {
            device_name,
            last_reading,
        }) => {
            match last_reading {
                Some(degrees) => {
                    println!("{device_name} handed over its last reading {degrees:.1}")
                }
                None => println!("{device_name} handed over no reading"),
            }
            Ok(Control::Continue(devices))
        }

        Incoming::Terminated(device_id) => {
            devices.retain(|device_name, device| {
                let gone = device.actor_id() == device_id;
                if gone {
                    println!("{device_name} is gone");
                }
                !gone
            });

            if devices.is_empty() {
                Ok(Control::Stop)
            } else {
                Ok(Control::Continue(devices))
            }
        }
    }
}

Right after spawning a device, the group calls context.watch(&device). From then on, when the device terminates, the group receives Incoming::Terminated with the device's ActorId. The group removes the device from its registry, and once the last device is gone it stops itself.

For Retire, the group needs the device's answer in its own mailbox, as a GroupMessage. context.reply_to creates such a ReplyTo: the closure turns the device's reply into a Retired message for the group. This is how actors ask each other without ever waiting.

The ordering guarantee

tellus/examples/tutorials/getting_started/04_watch.rs

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let system = ActorSystem::new(Group);
    let group = system.root();

    group.tell(GroupMessage::Record {
        device_name: "left",
        degrees: 21.5,
    });
    group.tell(GroupMessage::Record {
        device_name: "right",
        degrees: 19.1,
    });

    let last_reading = group
        .ask(Duration::from_secs(1), |reply_to| GroupMessage::Read {
            device_name: "left",
            reply_to,
        })
        .await
        .context("asking for the last reading")?;
    match last_reading {
        Some(degrees) => println!("Last reading of left: {degrees:.1}"),
        None => println!("No reading of left"),
    }

    group.tell(GroupMessage::Retire {
        device_name: "left",
    });
    group.tell(GroupMessage::Retire {
        device_name: "right",
    });

    system
        .terminated()
        .await
        .context("awaiting actor system termination")
}

Run it:

cargo run --quiet -p tellus --example getting_started_04_watch

Last reading of left: 21.5
right handed over its last reading 19.1
right is gone
left handed over its last reading 21.5
left is gone

The first line is the answer to the ask in main, which still reads a device through the group as in the previous step. After that, the two devices run concurrently, so left and right may appear in either order. But for each device, the handover always comes before "is gone". That is not luck. tellus guarantees that the terminated signal is ordered behind every message the terminated actor delivered to the watcher.

So receiving Terminated proves that you have seen every message from that actor you will ever see. The group can rely on this: when it forgets a device, no handover from it can still be on the way. You will use the same guarantee in the next step to avoid waiting for answers that will never come.

What you learned

context.watch turns another actor's termination into an Incoming::Terminated signal, and the signal arrives after all of that actor's messages. context.reply_to lets an actor receive a reply as an ordinary message.