//! A typed worker pool over process groups (`pg`).
//!
//! Workers enroll in a named group; the dispatcher reaches them through it.
//! Because the pool is homogeneous — every member is a `Worker` — the group's
//! typed reads (`pick_as` / `members_as`) and the `dispatch` combinator hand
//! members back as `Pid<Worker>`, so every send is an ordinary compile-checked
//! `send_to` rather than the untyped escape hatch. The untyped reads
//! (`members` / `pick` + `send_dyn`) stay available for identity-only use —
//! counting, logging, monitoring — where the message type isn't known.
//!
//! The pool drains itself: each worker retires on a sentinel and reports its
//! tally back, and the dispatcher waits the pool out before returning. (A pool
//! left running would be stopped anyway when the root actor exits, but draining
//! explicitly keeps the example deterministic.)

use smarm::{
    channel, dispatch, join, members, members_as, pick, pick_as, run, send_dyn, send_to,
    spawn_addr, Addressable, Pid,
};

/// The pool's worker actor. One message type, carried by `Pid<Worker>`.
struct Worker;
impl Addressable for Worker {
    type Msg = Job;
}

/// A unit of work, or the sentinel that retires a worker.
enum Job {
    Task { id: u64 },
    Retire,
}

const POOL: &str = "pool";
const WORKERS: u64 = 4;

fn main() {
    run(|| {
        // Mint four typed workers and enroll them. `spawn_addr` yields a
        // `Pid<Worker>`; `join` takes a typed pid directly, erasing internally.
        // Each worker reports how many tasks it handled over a shared channel,
        // so the dispatcher can wait the pool out at the end.
        let (done_tx, done_rx) = channel::<(u64, u64)>();
        for id in 0..WORKERS {
            let done_tx = done_tx.clone();
            let w: Pid<Worker> = spawn_addr::<Worker>(move |rx| {
                let mut handled = 0;
                while let Ok(job) = rx.recv() {
                    match job {
                        Job::Task { id: job } => {
                            println!("worker {id} handling job {job}");
                            handled += 1;
                        }
                        Job::Retire => break,
                    }
                }
                done_tx.send((id, handled)).unwrap();
            });
            join(POOL, w);
        }
        drop(done_tx); // from here only the workers hold senders

        // Pick one live member and hand it a job — typed end to end.
        if let Some(w) = pick_as::<Worker>(POOL) {
            send_to(w, Job::Task { id: 1 }).unwrap();
        }

        // `dispatch` rolls pick-a-live-member-and-send into one call, returning
        // the member it reached (or handing the job back if the pool is empty).
        let _ = dispatch::<Worker>(POOL, Job::Task { id: 2 });

        // Fan a job out to the whole pool — typed, so each send is `send_to`.
        for w in members_as::<Worker>(POOL) {
            let _ = send_to(w, Job::Task { id: 3 });
        }

        // The untyped reads stay available for identity-only use. To message a
        // member reached this way, the explicit `send_dyn` escape hatch names
        // the message type.
        println!("pool size: {}", members(POOL).len());
        if let Some(any) = pick(POOL) {
            let _ = send_dyn::<Job>(any, Job::Task { id: 4 });
        }

        // Retire every worker, then drain their tallies so the program winds
        // down on its own.
        for w in members_as::<Worker>(POOL) {
            let _ = send_to(w, Job::Retire);
        }
        for _ in 0..WORKERS {
            if let Ok((id, handled)) = done_rx.recv() {
                println!("worker {id} retired after {handled} jobs");
            }
        }
    });
}