Use actix for update_listener

src/update_listener.rs

@@ -3,37 +3,38 @@ use crate::{
     program_runner::{ProgramEvent, ProgramEventRecv},
     section_runner::{SecRunnerState, SecRunnerStateRecv, SectionEvent, SectionEventRecv},
 };
-use tokio::{
+use actix::{fut::wrap_future, Actor, ActorContext, Addr, AsyncContext, Handler, StreamHandler};
-    select,
+use tokio::sync::broadcast;
-    sync::{broadcast, oneshot},
+use tracing::{trace, warn};
-    task::JoinHandle,
-};
-use tracing::{trace_span, warn};
-use tracing_futures::Instrument;
 
-struct UpdateListenerTask {
+struct UpdateListenerActor {
-    section_events: SectionEventRecv,
-    program_events: ProgramEventRecv,
-    sec_runner_state: SecRunnerStateRecv,
     mqtt_interface: MqttInterface,
-    running: bool,
 }
 
-impl UpdateListenerTask {
+impl Actor for UpdateListenerActor {
-    async fn handle_section_event(
+    type Context = actix::Context<Self>;
+
+    fn started(&mut self, _ctx: &mut Self::Context) {
+        trace!("starting UpdateListener");
+    }
+
+    fn stopped(&mut self, _ctx: &mut Self::Context) {
+        trace!("stopped UpdateListener")
+    }
+}
+
+impl StreamHandler<Result<SectionEvent, broadcast::RecvError>> for UpdateListenerActor {
+    fn handle(
         &mut self,
         event: Result<SectionEvent, broadcast::RecvError>,
-    ) -> eyre::Result<()> {
+        ctx: &mut Self::Context,
+    ) {
         let event = match event {
             Ok(ev) => ev,
-            Err(broadcast::RecvError::Closed) => {
+            Err(broadcast::RecvError::Closed) => unreachable!(),
-                warn!("section events channel closed");
-                self.running = false;
-                return Ok(());
-            }
             Err(broadcast::RecvError::Lagged(n)) => {
                 warn!("section events lagged by {}", n);
-                return Ok(());
+                return;
             }
         };
         if let Some((sec_id, state)) = match event {
@@ -45,94 +46,116 @@ impl UpdateListenerTask {
             | SectionEvent::RunCancel(_, sec) => Some((sec.id, false)),
             SectionEvent::RunnerPause | SectionEvent::RunnerUnpause => None,
         } {
-            self.mqtt_interface
+            let mut mqtt_interface = self.mqtt_interface.clone();
-                .publish_section_state(sec_id, state)
+            let fut = async move {
-                .await?;
+                if let Err(err) = mqtt_interface.publish_section_state(sec_id, state).await {
+                    warn!("could not publish section state: {}", err);
+                }
+            };
+            ctx.spawn(wrap_future(fut));
         }
-        Ok(())
     }
+}
 
-    async fn handle_program_event(
+impl StreamHandler<Result<ProgramEvent, broadcast::RecvError>> for UpdateListenerActor {
+    fn handle(
         &mut self,
         event: Result<ProgramEvent, broadcast::RecvError>,
-    ) -> eyre::Result<()> {
+        ctx: &mut Self::Context,
+    ) {
         let event = match event {
             Ok(ev) => ev,
-            Err(broadcast::RecvError::Closed) => {
+            Err(broadcast::RecvError::Closed) => unreachable!(),
-                warn!("program events channel closed");
-                self.running = false;
-                return Ok(());
-            }
             Err(broadcast::RecvError::Lagged(n)) => {
                 warn!("program events lagged by {}", n);
-                return Ok(());
+                return;
             }
         };
         let (program_id, running) = match event {
             ProgramEvent::RunStart(prog) => (prog.id, true),
             ProgramEvent::RunFinish(prog) | ProgramEvent::RunCancel(prog) => (prog.id, false),
         };
-        self.mqtt_interface
+        let mut mqtt_interface = self.mqtt_interface.clone();
+        let fut = async move {
+            if let Err(err) = mqtt_interface
                 .publish_program_running(program_id, running)
-            .await?;
+                .await
-        Ok(())
+            {
-    }
+                warn!("could not publish program running: {}", err);
-
-    async fn handle_sec_runner_state(&mut self, state: Option<SecRunnerState>) -> eyre::Result<()> {
-        let state = match state {
-            Some(state) => state,
-            None => {
-                warn!("section runner events channel closed");
-                self.running = false;
-                return Ok(());
             }
         };
-        self.mqtt_interface.publish_section_runner(&state).await?;
+        ctx.spawn(wrap_future(fut));
-        Ok(())
     }
+}
 
-    async fn run_impl(&mut self) -> eyre::Result<()> {
+impl StreamHandler<SecRunnerState> for UpdateListenerActor {
-        while self.running {
+    fn handle(&mut self, state: SecRunnerState, ctx: &mut Self::Context) {
-            select! {
+        let mut mqtt_interface = self.mqtt_interface.clone();
-                section_event = self.section_events.recv() => {
+        let fut = async move {
-                    self.handle_section_event(section_event).await?
+            if let Err(err) = mqtt_interface.publish_section_runner(&state).await {
-                }
+                warn!("could not publish section runner: {}", err);
-                program_event = self.program_events.recv() => {
-                    self.handle_program_event(program_event).await?
-                }
-                sec_runner_state = self.sec_runner_state.recv() => {
-                    self.handle_sec_runner_state(sec_runner_state).await?
             }
         };
+        ctx.spawn(wrap_future(fut));
     }
-        Ok(())
+}
-    }
 
-    async fn run_or_quit(mut self, mut quit_rx: oneshot::Receiver<()>) -> eyre::Result<()> {
+#[derive(actix::Message)]
-        select! {
+#[rtype(result = "()")]
-            _ = &mut quit_rx => {
+struct Quit;
-                self.running = false;
+
-                Ok(())
+impl Handler<Quit> for UpdateListenerActor {
-            }
+    type Result = ();
-            res = self.run_impl() => {
+
-                res
+    fn handle(&mut self, _msg: Quit, ctx: &mut Self::Context) -> Self::Result {
+        ctx.stop();
     }
+}
+
+trait Listenable<A>: Send
+where
+    A: Actor,
+{
+    fn listen(self, ctx: &mut A::Context);
+}
+
+#[derive(actix::Message)]
+#[rtype(result = "()")]
+struct Listen<L>(L)
+where
+    L: Listenable<UpdateListenerActor>;
+
+impl<L> Handler<Listen<L>> for UpdateListenerActor
+where
+    L: Listenable<Self>,
+{
+    type Result = ();
+
+    fn handle(&mut self, msg: Listen<L>, ctx: &mut Self::Context) -> Self::Result {
+        msg.0.listen(ctx);
     }
+}
+
+impl Listenable<UpdateListenerActor> for SectionEventRecv {
+    fn listen(self, ctx: &mut <UpdateListenerActor as Actor>::Context) {
+        ctx.add_stream(self);
     }
+}
 
-    async fn run(self, quit_rx: oneshot::Receiver<()>) {
+impl Listenable<UpdateListenerActor> for ProgramEventRecv {
-        let span = trace_span!("update_listener task");
+    fn listen(self, ctx: &mut <UpdateListenerActor as Actor>::Context) {
+        ctx.add_stream(self);
+    }
+}
 
-        self.run_or_quit(quit_rx)
+impl Listenable<UpdateListenerActor> for SecRunnerStateRecv {
-            .instrument(span)
+    fn listen(self, ctx: &mut <UpdateListenerActor as Actor>::Context) {
-            .await
+        ctx.add_stream(self);
-            .expect("error in UpdateListenerTask");
     }
 }
 
 pub struct UpdateListener {
-    quit_tx: oneshot::Sender<()>,
+    addr: Addr<UpdateListenerActor>,
-    join_handle: JoinHandle<()>,
 }
 
 impl UpdateListener {
@@ -142,24 +165,22 @@ impl UpdateListener {
         sec_runner_state: SecRunnerStateRecv,
         mqtt_interface: MqttInterface,
     ) -> Self {
-        let task = UpdateListenerTask {
+        let addr = UpdateListenerActor { mqtt_interface }.start();
-            section_events,
+        let mut l = Self { addr };
-            program_events,
+        l.listen(section_events);
-            sec_runner_state,
+        l.listen(program_events);
-            mqtt_interface,
+        l.listen(sec_runner_state);
-            running: true,
+        l
-        };
-        let (quit_tx, quit_rx) = oneshot::channel();
-        let join_handle = tokio::spawn(task.run(quit_rx));
-        Self {
-            quit_tx,
-            join_handle,
     }
+
+    fn listen<L: 'static>(&mut self, listener: L)
+    where
+        L: Listenable<UpdateListenerActor>,
+    {
+        self.addr.do_send(Listen(listener));
     }
 
     pub async fn quit(self) -> eyre::Result<()> {
-        let _ = self.quit_tx.send(());
+        Ok(self.addr.send(Quit).await?)
-        self.join_handle.await?;
-        Ok(())
     }
 }