1 files changed, 25 insertions, 3 deletions
diff --git a/pkgs/by-name/mp/mpdpopm/src/lib.rs b/pkgs/by-name/mp/mpdpopm/src/lib.rs
index d5db57b4..cc2765dc 100644
--- a/pkgs/by-name/mp/mpdpopm/src/lib.rs
+++ b/pkgs/by-name/mp/mpdpopm/src/lib.rs
@@ -34,7 +34,9 @@
 
 pub mod clients;
 pub mod config;
+pub mod dj;
 pub mod filters_ast;
+pub mod messanges;
 pub mod playcounts;
 pub mod storage;
 pub mod vars;
@@ -51,6 +53,7 @@ pub mod filters {
 use crate::{
     clients::{Client, IdleClient, IdleSubSystem},
     config::{Config, Connection},
+    messanges::MessageQueue,
     playcounts::PlayState,
 };
 
@@ -90,6 +93,8 @@ pub async fn mpdpopm(cfg: Config) -> std::result::Result<(), Error> {
             .context("Failed to connect to TCP idle client")?,
     };
 
+    let mut mqueue = MessageQueue::new();
+
     idle_client
         .subscribe(&cfg.commands_chan)
         .await
@@ -106,6 +111,7 @@ pub async fn mpdpopm(cfg: Config) -> std::result::Result<(), Error> {
     pin_mut!(ctrl_c, sighup, sigkill, tick);
 
     let mut done = false;
+    let mut msg_check_needed = false;
     while !done {
         debug!("selecting...");
         {
@@ -132,7 +138,7 @@ pub async fn mpdpopm(cfg: Config) -> std::result::Result<(), Error> {
                         tick.set(sleep(Duration::from_millis(cfg.poll_interval_ms)).fuse());
                         state.update(&mut client)
                                 .await
-                                .context("PlayState update failed")?
+                                .context("PlayState update failed")?;
                     },
                     res = idle => match res {
                         Ok(subsys) => {
@@ -140,9 +146,14 @@ pub async fn mpdpopm(cfg: Config) -> std::result::Result<(), Error> {
                             if subsys == IdleSubSystem::Player {
                                 state.update(&mut client)
                                     .await
-                                    .context("PlayState update failed")?
+                                    .context("PlayState update failed")?;
+
+                                mqueue
+                                    .advance_dj(&mut client)
+                                    .await
+                                    .context("MessageQueue tick failed")?;
                             } else if subsys == IdleSubSystem::Message {
-                                error!("Message handling is not supported!");
+                                msg_check_needed = true;
                             }
                             break;
                         },
@@ -155,6 +166,17 @@ pub async fn mpdpopm(cfg: Config) -> std::result::Result<(), Error> {
                 }
             }
         }
+
+        if msg_check_needed {
+            msg_check_needed = false;
+
+            // Check for any messages that have come in; if there's an error there's not a lot we
+            // can do about it (suppose some client fat-fingers a command name, e.g.)-- just log it
+            // & move on.
+            if let Err(err) = mqueue.check_messages(&mut client, &mut idle_client).await {
+                error!("Error while processing messages: {err:#?}");
+            }
+        }
     }
 
     info!("mpdpopm exiting.");