tellus 0.2.1

//! In this example the root actor scatters a workload across worker actors and gathers their
//! partial results: each request carries a `ReplyTo` created via `ActorContext::reply_to`, so each
//! worker sums up its shard, replies the result and stops. The root watches its workers; a
//! terminated signal is ordered behind everything the terminated actor has delivered to the
//! watcher, so receiving one proves that worker's partial result has already been added.
//! Hence, once all workers have terminated, the root can print the total and stop, which terminates
//! the actor system.
//!
//! The total is printed to stdout and tellus logs to stderr; the log level is configured via
//! `RUST_LOG`, e.g. `RUST_LOG=tellus=debug cargo run --quiet -p tellus --example scatter_gather`.

use anyhow::Context;
use std::{convert::Infallible, io, ops::Range};
use tellus::{Actor, ActorContext, ActorSystem, Control, Incoming, ReplyTo};
use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitExt};

const SHARDS: [Range<u64>; 4] = [1..26, 26..51, 51..76, 76..101];

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    init_tracing();

    let system = ActorSystem::new(Gatherer);

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

fn init_tracing() {
    tracing_subscriber::registry()
        .with(EnvFilter::from_default_env())
        .with(
            tracing_subscriber::fmt::layer()
                .json()
                .flatten_event(true)
                .with_writer(io::stderr),
        )
        .init();
}

struct Gatherer;

impl Actor for Gatherer {
    type Message = Partial;
    type State = Gathering;
    type Error = Infallible;

    fn init(&self, context: &ActorContext<Self::Message>) -> Result<Self::State, Self::Error> {
        for shard in SHARDS {
            let worker = context.spawn(Worker);
            context.watch(&worker);

            worker.tell(Compute {
                shard,
                reply_to: context.reply_to(Partial),
            });
        }

        Ok(Gathering {
            remaining: SHARDS.len(),
            total: 0,
        })
    }

    fn receive(
        &self,
        _: &ActorContext<Self::Message>,
        incoming: Incoming<Self::Message>,
        state: Self::State,
    ) -> Result<Control<Self::State>, Self::Error> {
        match incoming {
            Incoming::Message(Partial(sum)) => Ok(Control::Continue(Gathering {
                total: state.total + sum,
                ..state
            })),

            // The partial result of the terminated worker has already been added.
            Incoming::Terminated(_) => {
                let remaining = state.remaining - 1;

                if remaining > 0 {
                    Ok(Control::Continue(Gathering { remaining, ..state }))
                } else {
                    println!("## Total is: {}", state.total);
                    Ok(Control::Stop)
                }
            }
        }
    }
}

struct Partial(u64);

struct Gathering {
    remaining: usize,
    total: u64,
}

struct Worker;

impl Actor for Worker {
    type Message = Compute;
    type State = ();
    type Error = Infallible;

    fn init(&self, _: &ActorContext<Self::Message>) -> Result<Self::State, Self::Error> {
        Ok(())
    }

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

        let (start, end) = (shard.start, shard.end);
        let sum = shard.sum::<u64>();
        println!("## Shard {start}..{end} sums up to: {sum}");
        reply_to.reply(sum);

        Ok(Control::Stop)
    }
}

struct Compute {
    shard: Range<u64>,
    reply_to: ReplyTo<u64>,
}