sprinklers_rs/sprinklers_actors/src/program_runner.rs

use crate::zone_runner::{
    Error as ZoneRunnerError, ZoneEvent, ZoneEventRecv, ZoneRunHandle, ZoneRunner,
};
use sprinklers_core::model::{ProgramId, ProgramRef, Programs, Zones};

use actix::{
    Actor, ActorContext, ActorFuture, ActorStream, Addr, AsyncContext, Handler, Message,
    MessageResult, SpawnHandle, StreamHandler, WrapFuture,
};
use std::collections::VecDeque;
use thiserror::Error;
use tokio::{
    sync::{broadcast, watch},
    time::{delay_queue, DelayQueue},
};
use tracing::{debug, error, trace, warn};

#[derive(Clone, Debug)]
pub enum ProgramEvent {
    RunStart(ProgramRef),
    RunFinish(ProgramRef),
    RunCancel(ProgramRef),
    NextRun(ProgramRef, chrono::DateTime<chrono::Local>),
}

pub type ProgramEventRecv = broadcast::Receiver<ProgramEvent>;
type ProgramEventSend = broadcast::Sender<ProgramEvent>;

const EVENT_CAPACITY: usize = 8;

#[derive(Clone, Debug, PartialEq)]
enum RunState {
    Waiting,
    Running,
    Finished,
    Cancelled,
}

#[derive(Debug)]
struct ProgRun {
    program: ProgramRef,
    state: RunState,
    zone_run_handles: Vec<ZoneRunHandle>,
}

impl ProgRun {
    fn new(program: ProgramRef) -> Self {
        Self {
            program,
            state: RunState::Waiting,
            zone_run_handles: Vec::new(),
        }
    }
}

type RunQueue = VecDeque<ProgRun>;

struct ProgramRunnerInner {
    zone_runner: ZoneRunner,
    zones: Zones,
    programs: Programs,
    event_send: Option<ProgramEventSend>,
    schedule_run_fut: Option<SpawnHandle>,
}

impl ProgramRunnerInner {
    fn send_event(&mut self, event: ProgramEvent) {
        if let Some(event_send) = &mut self.event_send {
            match event_send.send(event) {
                Ok(_) => {}
                Err(_closed) => {
                    self.event_send = None;
                }
            }
        }
    }

    fn subscribe_event(&mut self) -> ProgramEventRecv {
        match &mut self.event_send {
            Some(event_send) => event_send.subscribe(),
            None => {
                let (event_send, event_recv) = broadcast::channel(EVENT_CAPACITY);
                self.event_send = Some(event_send);
                event_recv
            }
        }
    }

    fn start_program_run(&mut self, run: &mut ProgRun) {
        if run.state != RunState::Waiting {
            warn!(
                program_id = run.program.id,
                "cannot run program which is already running"
            );
            return;
        }
        let sequence: Vec<_> = run
            .program
            .sequence
            .iter()
            .filter_map(|item| match self.zones.get(&item.zone_id) {
                Some(zone) => Some((zone.clone(), item.duration)),
                None => {
                    warn!(
                        program_id = run.program.id,
                        zone_id = item.zone_id,
                        "trying to run program with nonexistant zone"
                    );
                    None
                }
            })
            .collect();
        if sequence.is_empty() {
            warn!(program_id = run.program.id, "program has no valid zones");
            run.state = RunState::Finished;
            self.send_event(ProgramEvent::RunStart(run.program.clone()));
            self.send_event(ProgramEvent::RunFinish(run.program.clone()));
            return;
        }
        run.zone_run_handles.reserve(sequence.len());
        for (zone, duration) in sequence {
            let handle = self.zone_runner.do_queue_run(zone, duration);
            run.zone_run_handles.push(handle);
        }
        run.state = RunState::Running;
        self.send_event(ProgramEvent::RunStart(run.program.clone()));
        debug!(program_id = run.program.id, "started running program");
    }

    fn cancel_program_run(&mut self, run: &mut ProgRun) {
        for handle in run.zone_run_handles.drain(..) {
            self.zone_runner.do_cancel_run(handle);
        }
        debug!(program_id = run.program.id, "program run is cancelled");
        self.send_event(ProgramEvent::RunCancel(run.program.clone()));
    }

    fn update_schedules(&mut self, ctx: &mut actix::Context<ProgramRunnerActor>) {
        let mut scheduled_run_queue = DelayQueue::with_capacity(self.programs.len());
        for (_, prog) in self.programs.clone() {
            if !prog.enabled {
                // TODO: send NextRun(prog, None) so nextRun will be updated
                continue;
            }
            let ref_time = chrono::Local::now();
            let next_run = match prog.schedule.next_run_after(&ref_time) {
                Some(next_run) => next_run,
                None => continue,
            };
            let delay = (next_run - ref_time).to_std().unwrap();
            trace!("will run program in {:?}", delay);
            scheduled_run_queue.insert(prog.clone(), delay);
            self.send_event(ProgramEvent::NextRun(prog, next_run));
        }
        let fut = actix::fut::wrap_stream(scheduled_run_queue)
            .map(|item, act: &mut ProgramRunnerActor, ctx| act.handle_scheduled_run(item, ctx))
            .finish();
        let handle = ctx.spawn(fut);
        if let Some(old_handle) = self.schedule_run_fut.replace(handle) {
            ctx.cancel_future(old_handle);
        }
    }
}

struct ProgramRunnerActor {
    inner: ProgramRunnerInner,
    run_queue: RunQueue,
}

impl Actor for ProgramRunnerActor {
    type Context = actix::Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        trace!("subscribing to ZoneRunner events");
        let subscribe_fut = self.inner.zone_runner.subscribe().into_actor(self).map(
            |zone_events: Result<ZoneEventRecv, ZoneRunnerError>,
             _act: &mut ProgramRunnerActor,
             ctx: &mut Self::Context| {
                match zone_events {
                    Ok(zone_events) => {
                        ctx.add_stream(zone_events.into_stream());
                    }
                    Err(err) => warn!("failed to subscribe to ZoneRunner events: {}", err),
                }
            },
        );
        ctx.wait(subscribe_fut);
        trace!("program_runner starting");
    }

    fn stopped(&mut self, _ctx: &mut Self::Context) {
        trace!("program_runner stopped");
    }
}

impl StreamHandler<Result<ZoneEvent, broadcast::RecvError>> for ProgramRunnerActor {
    fn handle(&mut self, item: Result<ZoneEvent, broadcast::RecvError>, ctx: &mut Self::Context) {
        let zone_event = match item {
            Ok(e) => e,
            Err(err) => {
                warn!("failed to receive zone event: {}", err);
                return;
            }
        };
        #[allow(clippy::single_match)]
        match zone_event {
            ZoneEvent::RunFinish(finished_run, _) => {
                self.handle_finished_run(finished_run, ctx);
            }
            _ => {}
        }
    }
}

#[derive(Message)]
#[rtype(result = "()")]
struct Quit;

impl Handler<Quit> for ProgramRunnerActor {
    type Result = ();
    fn handle(&mut self, _msg: Quit, ctx: &mut Self::Context) -> Self::Result {
        ctx.stop();
    }
}

#[derive(Message)]
#[rtype(result = "ProgramEventRecv")]
struct Subscribe;

impl Handler<Subscribe> for ProgramRunnerActor {
    type Result = MessageResult<Subscribe>;
    fn handle(&mut self, _msg: Subscribe, _ctx: &mut Self::Context) -> Self::Result {
        let event_recv = self.inner.subscribe_event();
        MessageResult(event_recv)
    }
}

#[derive(Message)]
#[rtype(result = "()")]
struct UpdateZones(Zones);

impl Handler<UpdateZones> for ProgramRunnerActor {
    type Result = ();
    fn handle(&mut self, msg: UpdateZones, _ctx: &mut Self::Context) -> Self::Result {
        trace!("updating zones");
        let UpdateZones(new_zones) = msg;
        self.inner.zones = new_zones;
    }
}

#[derive(Message)]
#[rtype(result = "()")]
struct UpdatePrograms(Programs);

impl Handler<UpdatePrograms> for ProgramRunnerActor {
    type Result = ();
    fn handle(&mut self, msg: UpdatePrograms, ctx: &mut Self::Context) -> Self::Result {
        trace!("updating programs");
        let UpdatePrograms(new_programs) = msg;
        self.inner.programs = new_programs;
        self.inner.update_schedules(ctx);
    }
}

#[derive(Message)]
#[rtype(result = "Result<ProgramRef>")]
struct RunProgramId(ProgramId);

impl Handler<RunProgramId> for ProgramRunnerActor {
    type Result = Result<ProgramRef>;
    fn handle(&mut self, msg: RunProgramId, ctx: &mut Self::Context) -> Self::Result {
        let RunProgramId(program_id) = msg;
        let program = match self.inner.programs.get(&program_id) {
            Some(program) => program.clone(),
            None => {
                trace!(program_id, "trying to run non-existant program");
                return Err(Error::InvalidProgramId(program_id));
            }
        };
        self.run_queue.push_back(ProgRun::new(program.clone()));
        ctx.notify(Process);
        Ok(program)
    }
}

#[derive(Message)]
#[rtype(result = "()")]
struct RunProgram(ProgramRef);

impl Handler<RunProgram> for ProgramRunnerActor {
    type Result = ();
    fn handle(&mut self, msg: RunProgram, ctx: &mut Self::Context) -> Self::Result {
        let RunProgram(program) = msg;
        self.run_queue.push_back(ProgRun::new(program));
        ctx.notify(Process);
    }
}

#[derive(Message)]
#[rtype(result = "Option<ProgramRef>")]
struct CancelProgram(ProgramId);

impl Handler<CancelProgram> for ProgramRunnerActor {
    type Result = Option<ProgramRef>;
    fn handle(&mut self, msg: CancelProgram, ctx: &mut Self::Context) -> Self::Result {
        let CancelProgram(program_id) = msg;
        let mut cancelled = None;
        for run in self.run_queue.iter_mut() {
            if run.program.id == program_id {
                run.state = RunState::Cancelled;
                cancelled = Some(run.program.clone());
            }
        }
        ctx.notify(Process);
        cancelled
    }
}

impl StreamHandler<Zones> for ProgramRunnerActor {
    fn handle(&mut self, item: Zones, ctx: &mut Self::Context) {
        ctx.notify(UpdateZones(item))
    }
}

#[derive(Message)]
#[rtype(result = "()")]
struct ListenZones(watch::Receiver<Zones>);

impl Handler<ListenZones> for ProgramRunnerActor {
    type Result = ();

    fn handle(&mut self, msg: ListenZones, ctx: &mut Self::Context) -> Self::Result {
        ctx.add_stream(msg.0);
    }
}

impl StreamHandler<Programs> for ProgramRunnerActor {
    fn handle(&mut self, item: Programs, ctx: &mut Self::Context) {
        ctx.notify(UpdatePrograms(item))
    }
}

#[derive(Message)]
#[rtype(result = "()")]
struct ListenPrograms(watch::Receiver<Programs>);

impl Handler<ListenPrograms> for ProgramRunnerActor {
    type Result = ();

    fn handle(&mut self, msg: ListenPrograms, ctx: &mut Self::Context) -> Self::Result {
        ctx.add_stream(msg.0);
    }
}

#[derive(Message)]
#[rtype(result = "()")]
struct Process;

impl Handler<Process> for ProgramRunnerActor {
    type Result = ();
    fn handle(&mut self, _msg: Process, _ctx: &mut Self::Context) -> Self::Result {
        while let Some(current_run) = self.run_queue.front_mut() {
            let run_finished = match current_run.state {
                RunState::Waiting => {
                    self.inner.start_program_run(current_run);
                    false
                }
                RunState::Running => false,
                RunState::Finished => true,
                RunState::Cancelled => {
                    self.inner.cancel_program_run(current_run);
                    true
                }
            };
            if run_finished {
                self.run_queue.pop_front();
            } else {
                break;
            }
        }
    }
}

#[derive(Message)]
#[rtype(result = "()")]
struct UpdateSchedules;

impl Handler<UpdateSchedules> for ProgramRunnerActor {
    type Result = ();
    fn handle(&mut self, _msg: UpdateSchedules, ctx: &mut Self::Context) -> Self::Result {
        self.inner.update_schedules(ctx);
    }
}

impl ProgramRunnerActor {
    fn new(zone_runner: ZoneRunner) -> Self {
        Self {
            inner: ProgramRunnerInner {
                zone_runner,
                zones: Zones::new(),
                programs: Programs::new(),
                event_send: None,
                schedule_run_fut: None,
            },
            run_queue: RunQueue::new(),
        }
    }

    fn handle_finished_run(
        &mut self,
        finished_run: ZoneRunHandle,
        ctx: &mut <ProgramRunnerActor as Actor>::Context,
    ) -> Option<()> {
        let current_run = self.run_queue.front_mut()?;
        let last_run_handle = current_run.zone_run_handles.last()?;
        if finished_run == *last_run_handle {
            current_run.state = RunState::Finished;
            debug!(
                program_id = current_run.program.id,
                "finished running program"
            );
            self.inner
                .send_event(ProgramEvent::RunFinish(current_run.program.clone()));
            ctx.notify(Process);
        }
        Some(())
    }

    fn handle_scheduled_run(
        &mut self,
        item: Result<delay_queue::Expired<ProgramRef>, tokio::time::Error>,
        ctx: &mut <ProgramRunnerActor as Actor>::Context,
    ) {
        let program = match item {
            Ok(expired) => expired.into_inner(),
            Err(err) => {
                error!("tokio time error: {}", err);
                return;
            }
        };
        trace!(program_id = program.id, "schedule expired");
        self.run_queue.push_back(ProgRun::new(program));
        ctx.notify(Process);
        ctx.notify(UpdateSchedules);
    }
}

#[derive(Debug, Clone, Error)]
pub enum Error {
    #[error("mailbox error: {0}")]
    Mailbox(
        #[from]
        #[source]
        actix::MailboxError,
    ),
    #[error("no such program id: {0}")]
    InvalidProgramId(ProgramId),
}

pub type Result<T, E = Error> = std::result::Result<T, E>;

#[derive(Clone)]
pub struct ProgramRunner {
    addr: Addr<ProgramRunnerActor>,
}

#[allow(dead_code)]
impl ProgramRunner {
    pub fn new(zone_runner: ZoneRunner) -> Self {
        let addr = ProgramRunnerActor::new(zone_runner).start();
        Self { addr }
    }

    pub async fn quit(&mut self) -> Result<()> {
        self.addr.send(Quit).await?;
        Ok(())
    }

    pub async fn update_zones(&mut self, new_zones: Zones) -> Result<()> {
        self.addr
            .send(UpdateZones(new_zones))
            .await
            .map_err(Error::from)
    }

    pub async fn update_programs(&mut self, new_programs: Programs) -> Result<()> {
        self.addr
            .send(UpdatePrograms(new_programs))
            .await
            .map_err(Error::from)
    }

    pub async fn run_program_id(&mut self, program_id: ProgramId) -> Result<ProgramRef> {
        self.addr
            .send(RunProgramId(program_id))
            .await
            .map_err(Error::from)?
    }

    pub async fn run_program(&mut self, program: ProgramRef) -> Result<()> {
        self.addr
            .send(RunProgram(program))
            .await
            .map_err(Error::from)
    }

    pub async fn cancel_program(&mut self, program_id: ProgramId) -> Result<Option<ProgramRef>> {
        self.addr
            .send(CancelProgram(program_id))
            .await
            .map_err(Error::from)
    }

    pub async fn subscribe(&mut self) -> Result<ProgramEventRecv> {
        let event_recv = self.addr.send(Subscribe).await?;
        Ok(event_recv)
    }

    pub fn listen_zones(&mut self, zones_watch: watch::Receiver<Zones>) {
        // TODO: should this adopt a similar pattern to update_listener?
        self.addr.do_send(ListenZones(zones_watch))
    }

    pub fn listen_programs(&mut self, programs_watch: watch::Receiver<Programs>) {
        self.addr.do_send(ListenPrograms(programs_watch))
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use crate::trace_listeners::{EventListener, Filters};
    use sprinklers_core::{
        model::{Program, ProgramItem, Zone},
        schedule::{every_day, DateTimeBound, Schedule},
        zone_interface::{MockZoneInterface, ZoneInterface},
    };

    use assert_matches::assert_matches;
    use im::ordmap;
    use std::{sync::Arc, time::Duration};
    use tokio::task::yield_now;
    use tracing_subscriber::prelude::*;

    #[actix_rt::test]
    async fn test_quit() {
        let quit_msg = EventListener::new(
            Filters::new()
                .target("sprinklers_actors::program_runner")
                .message("program_runner stopped"),
        );
        let subscriber = tracing_subscriber::registry().with(quit_msg.clone());
        let _sub = tracing::subscriber::set_default(subscriber);

        let interface = MockZoneInterface::new(6);
        let mut zone_runner = ZoneRunner::new(Arc::new(interface));
        let mut runner = ProgramRunner::new(zone_runner.clone());
        yield_now().await;
        runner.quit().await.unwrap();
        zone_runner.quit().await.unwrap();
        yield_now().await;

        assert_eq!(quit_msg.get_count(), 1);
    }

    fn make_zones_and_runner() -> (Zones, ZoneRunner, Arc<MockZoneInterface>) {
        let interface = Arc::new(MockZoneInterface::new(2));
        let zones: Zones = ordmap![
            1 => Zone {
                id: 1,
                name: "Zone 1".into(),
                interface_id: 0,
            }.into(),
            2 => Zone {
                id: 2,
                name: "Zone 2".into(),
                interface_id: 1,
            }.into()
        ];
        let zone_runner = ZoneRunner::new(interface.clone());
        (zones, zone_runner, interface)
    }

    fn make_program(num: ProgramId, sequence: Vec<ProgramItem>) -> ProgramRef {
        make_program_with_schedule(num, sequence, false, Schedule::default())
    }

    fn make_program_with_schedule(
        num: ProgramId,
        sequence: Vec<ProgramItem>,
        enabled: bool,
        schedule: Schedule,
    ) -> ProgramRef {
        Program {
            id: num,
            name: format!("Program {}", num),
            sequence,
            enabled,
            schedule,
        }
        .into()
    }

    #[actix_rt::test]
    async fn test_run_program() {
        let (zones, mut zone_runner, interface) = make_zones_and_runner();
        let mut zone_events = zone_runner.subscribe().await.unwrap();
        let mut runner = ProgramRunner::new(zone_runner.clone());
        let mut prog_events = runner.subscribe().await.unwrap();

        let program = make_program(
            1,
            vec![
                ProgramItem {
                    zone_id: 1,
                    duration: Duration::from_secs(10),
                },
                ProgramItem {
                    zone_id: 2,
                    duration: Duration::from_secs(10),
                },
            ],
        );

        runner.update_zones(zones.clone()).await.unwrap();

        runner.run_program(program).await.unwrap();
        yield_now().await;
        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunStart(prog))
            if prog.id == 1
        );
        assert_matches!(zone_events.try_recv(), Ok(ZoneEvent::RunStart(_, _)));
        assert_eq!(interface.get_zone_state(0), true);

        tokio::time::pause();
        assert_matches!(zone_events.recv().await, Ok(ZoneEvent::RunFinish(_, _)));
        assert_matches!(zone_events.recv().await, Ok(ZoneEvent::RunStart(_, _)));
        assert_eq!(interface.get_zone_state(0), false);
        assert_eq!(interface.get_zone_state(1), true);
        assert_matches!(zone_events.recv().await, Ok(ZoneEvent::RunFinish(_, _)));
        assert_matches!(prog_events.recv().await, Ok(ProgramEvent::RunFinish(_)));

        runner.quit().await.unwrap();
        zone_runner.quit().await.unwrap();
        yield_now().await;
    }

    #[actix_rt::test]
    async fn test_run_nonexistant_zone() {
        let (zones, mut zone_runner, _) = make_zones_and_runner();
        let mut runner = ProgramRunner::new(zone_runner.clone());
        let mut prog_events = runner.subscribe().await.unwrap();

        let program1 = make_program(
            1,
            vec![ProgramItem {
                zone_id: 3,
                duration: Duration::from_secs(10),
            }],
        );
        let program2 = make_program(
            2,
            vec![ProgramItem {
                zone_id: 1,
                duration: Duration::from_secs(10),
            }],
        );
        runner.update_zones(zones.clone()).await.unwrap();

        runner.run_program(program1).await.unwrap();
        yield_now().await;
        // Should immediately start and finish running program
        // due to nonexistant zone
        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunStart(prog))
            if prog.id == 1
        );
        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunFinish(prog))
            if prog.id == 1
        );

        runner.run_program(program2).await.unwrap();
        yield_now().await;
        // Should run right away since last program should be done
        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunStart(prog))
            if prog.id == 2
        );
        tokio::time::pause();
        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunFinish(prog))
            if prog.id == 2
        );

        runner.quit().await.unwrap();
        zone_runner.quit().await.unwrap();
    }

    #[actix_rt::test]
    async fn test_close_event_chan() {
        let (zones, mut zone_runner, _) = make_zones_and_runner();
        let mut runner = ProgramRunner::new(zone_runner.clone());
        let mut prog_events = runner.subscribe().await.unwrap();

        let program = make_program(1, vec![]);

        runner.update_zones(zones.clone()).await.unwrap();

        runner.run_program(program.clone()).await.unwrap();
        prog_events.recv().await.unwrap();
        prog_events.recv().await.unwrap();
        // Make sure it doesn't panic when the events channel is dropped
        drop(prog_events);
        yield_now().await;
        runner.run_program(program).await.unwrap();
        yield_now().await;

        runner.quit().await.unwrap();
        zone_runner.quit().await.unwrap();
    }

    #[actix_rt::test]
    async fn test_run_program_id() {
        let (zones, mut zone_runner, _) = make_zones_and_runner();
        let mut runner = ProgramRunner::new(zone_runner.clone());
        let mut prog_events = runner.subscribe().await.unwrap();

        let program1 = make_program(
            1,
            vec![ProgramItem {
                zone_id: 2,
                duration: Duration::from_secs(10),
            }],
        );
        let program2 = make_program(
            2,
            vec![ProgramItem {
                zone_id: 2,
                duration: Duration::from_secs(10),
            }],
        );
        let programs = ordmap![ 1 => program1, 2 => program2 ];

        runner.update_zones(zones.clone()).await.unwrap();
        runner.update_programs(programs).await.unwrap();

        // First try a non-existant program id
        assert_matches!(
            runner.run_program_id(3).await,
            Err(Error::InvalidProgramId(3))
        );

        runner.run_program_id(1).await.unwrap();
        yield_now().await;
        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunStart(prog))
            if prog.id == 1
        );

        tokio::time::pause();
        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunFinish(prog))
            if prog.id == 1
        );
        runner.run_program_id(1).await.unwrap();
        yield_now().await;
        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunStart(prog))
            if prog.id == 1
        );

        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunFinish(prog))
            if prog.id == 1
        );

        runner.quit().await.unwrap();
        zone_runner.quit().await.unwrap();
    }

    #[actix_rt::test]
    async fn test_queue_program() {
        let (zones, mut zone_runner, _) = make_zones_and_runner();
        let mut runner = ProgramRunner::new(zone_runner.clone());
        let mut prog_events = runner.subscribe().await.unwrap();

        let program1 = make_program(
            1,
            vec![ProgramItem {
                zone_id: 2,
                duration: Duration::from_secs(10),
            }],
        );
        let program2 = make_program(
            2,
            vec![ProgramItem {
                zone_id: 2,
                duration: Duration::from_secs(10),
            }],
        );
        let programs = ordmap![ 1 => program1, 2 => program2 ];

        runner.update_zones(zones.clone()).await.unwrap();
        runner.update_programs(programs).await.unwrap();

        runner.run_program_id(1).await.unwrap();
        runner.run_program_id(2).await.unwrap();
        tokio::time::pause();
        tokio::time::delay_for(Duration::from_secs(21)).await;

        assert_matches!(
            prog_events.try_recv(),
            Ok(ProgramEvent::RunStart(prog))
            if prog.id == 1
        );
        assert_matches!(
            prog_events.try_recv(),
            Ok(ProgramEvent::RunFinish(prog))
            if prog.id == 1
        );
        assert_matches!(
            prog_events.try_recv(),
            Ok(ProgramEvent::RunStart(prog))
            if prog.id == 2
        );
        assert_matches!(
            prog_events.try_recv(),
            Ok(ProgramEvent::RunFinish(prog))
            if prog.id == 2
        );

        runner.quit().await.unwrap();
        zone_runner.quit().await.unwrap();
    }

    #[actix_rt::test]
    async fn test_cancel_program() {
        let (zones, mut zone_runner, _) = make_zones_and_runner();
        let mut zone_events = zone_runner.subscribe().await.unwrap();
        let mut runner = ProgramRunner::new(zone_runner.clone());
        let mut prog_events = runner.subscribe().await.unwrap();

        let program = make_program(
            1,
            vec![
                ProgramItem {
                    zone_id: 1,
                    duration: Duration::from_secs(10),
                },
                ProgramItem {
                    zone_id: 2,
                    duration: Duration::from_secs(10),
                },
            ],
        );

        runner.update_zones(zones.clone()).await.unwrap();

        runner.run_program(program.clone()).await.unwrap();
        yield_now().await;
        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunStart(prog))
            if prog.id == 1
        );
        assert_matches!(zone_events.try_recv().unwrap(), ZoneEvent::RunStart(_, _));

        runner.cancel_program(program.id).await.unwrap();
        yield_now().await;
        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunCancel(prog))
            if prog.id == 1
        );
        assert_matches!(zone_events.recv().await, Ok(ZoneEvent::RunCancel(_, _)));

        runner.quit().await.unwrap();
        zone_runner.quit().await.unwrap();
    }

    #[actix_rt::test]
    async fn test_scheduled_run() {
        let (zones, mut zone_runner, _) = make_zones_and_runner();
        let mut runner = ProgramRunner::new(zone_runner.clone());
        let mut prog_events = runner.subscribe().await.unwrap();

        let make_programs = |num: ProgramId, enabled: bool| {
            let now = chrono::Local::now();
            let sched_time = now.time() + chrono::Duration::microseconds(1000);
            let schedule = Schedule::new(
                vec![sched_time],
                every_day(),
                DateTimeBound::None,
                DateTimeBound::None,
            );
            let program1 = make_program_with_schedule(
                num,
                vec![ProgramItem {
                    zone_id: 1,
                    duration: Duration::from_micros(100),
                }],
                enabled,
                schedule,
            );
            let programs = ordmap![ num => program1 ];
            programs
        };

        runner.update_zones(zones.clone()).await.unwrap();
        runner
            .update_programs(make_programs(1, false))
            .await
            .unwrap();

        // TODO: would use tokio::time::pause here but that doesn't effect chrono now
        // which is used for schedules
        // tokio::time::pause();
        tokio::time::delay_for(Duration::from_micros(1100)).await;
        // Should not run (is disabled)
        assert_matches!(prog_events.try_recv(), Err(broadcast::TryRecvError::Empty));

        runner
            .update_programs(make_programs(2, true))
            .await
            .unwrap();

        // Should run
        tokio::time::delay_for(Duration::from_micros(1100)).await;
        assert_matches!(prog_events.recv().await, Ok(ProgramEvent::NextRun(_, _)));
        assert_matches!(
            prog_events.recv().await,
            Ok(ProgramEvent::RunStart(prog))
            if prog.id == 2
        );

        runner.quit().await.unwrap();
        zone_runner.quit().await.unwrap();
    }

    #[actix_rt::test]
    async fn test_scheduled_run_twice() {
        let (zones, mut zone_runner, _) = make_zones_and_runner();
        let mut runner = ProgramRunner::new(zone_runner.clone());
        let mut prog_events = runner.subscribe().await.unwrap();

        let now = chrono::Local::now();
        let sched_time1 = now.time() + chrono::Duration::microseconds(1000);
        let sched_time2 = now.time() + chrono::Duration::microseconds(5000);
        let schedule = Schedule::new(
            vec![sched_time1, sched_time2],
            every_day(),
            DateTimeBound::None,
            DateTimeBound::None,
        );
        let program1 = make_program_with_schedule(
            1,
            vec![ProgramItem {
                zone_id: 1,
                duration: Duration::from_micros(10),
            }],
            true,
            schedule,
        );
        let programs = ordmap![ 1 => program1 ];

        runner.update_zones(zones.clone()).await.unwrap();
        runner.update_programs(programs).await.unwrap();

        let fut = async move {
            // TODO: would use tokio::time::pause here but that doesn't effect chrono now
            // which is used for schedules
            // tokio::time::pause();
            // Should run
            tokio::time::delay_for(Duration::from_micros(1100)).await;
            assert_matches!(prog_events.recv().await, Ok(ProgramEvent::NextRun(_, _)));
            assert_matches!(
                prog_events.recv().await,
                Ok(ProgramEvent::RunStart(prog))
                if prog.id == 1
            );
            assert_matches!(prog_events.recv().await, Ok(ProgramEvent::NextRun(_, _)));
            assert_matches!(
                prog_events.recv().await,
                Ok(ProgramEvent::RunFinish(prog))
                if prog.id == 1
            );

            // Should run again
            assert_matches!(
                prog_events.recv().await,
                Ok(ProgramEvent::RunStart(prog))
                if prog.id == 1
            );
            tokio::task::yield_now().await;
            assert_matches!(prog_events.recv().await, Ok(ProgramEvent::NextRun(_, _)));
            assert_matches!(
                prog_events.recv().await,
                Ok(ProgramEvent::RunFinish(prog))
                if prog.id == 1
            );
        };
        // TODO: this still sometimes fails
        tokio::time::timeout(Duration::from_micros(10000), fut)
            .await
            .unwrap();

        runner.quit().await.unwrap();
        zone_runner.quit().await.unwrap();
    }
}