From 0dd5cf8fe4ff893cdbfcfddd3c6b4be9d2dd8a81 Mon Sep 17 00:00:00 2001
From: Tom Alexander <tom@fizz.buzz>
Date: Mon, 21 Feb 2022 00:05:36 -0500
Subject: [PATCH] Notifying listeners but sending is not working.

---
 src/mpvctl/mpvctl.rs | 40 +++++++++++++++++++++++++++++++++-------
 1 file changed, 33 insertions(+), 7 deletions(-)

diff --git a/src/mpvctl/mpvctl.rs b/src/mpvctl/mpvctl.rs
index 1b7f2a3..e5d70b1 100644
--- a/src/mpvctl/mpvctl.rs
+++ b/src/mpvctl/mpvctl.rs
@@ -33,10 +33,18 @@ impl MpvCtl {
         let mut framed_read = MpvFramed::new(socket_read);
         let db: CommandDb = Arc::new(Mutex::new(HashMap::new()));
         let read_db_handle = db.clone();
+        let event_listeners = Arc::new(Mutex::new(HashMap::new()));
+        let read_event_listeners_handle = event_listeners.clone();
 
         tokio::spawn(async move {
             loop {
-                match Self::read_loop(&mut framed_read, &read_db_handle).await {
+                match Self::read_loop(
+                    &mut framed_read,
+                    &read_db_handle,
+                    &read_event_listeners_handle,
+                )
+                .await
+                {
                     Ok(_) => (),
                     Err(e) => {
                         eprintln!("Error in read loop: {}", e);
@@ -49,7 +57,7 @@ impl MpvCtl {
             socket: socket_write,
             next_request_id: 0,
             in_flight_requests: db,
-            event_listeners: Arc::new(Mutex::new(HashMap::new())),
+            event_listeners,
         })
     }
 
@@ -78,6 +86,7 @@ impl MpvCtl {
     async fn read_loop(
         framed_read: &mut MpvFramed,
         db: &CommandDb,
+        event_listeners: &EventDb,
     ) -> Result<(), Box<dyn std::error::Error>> {
         match framed_read.read_frame().await {
             Ok(Some(frame)) => {
@@ -88,7 +97,28 @@ impl MpvCtl {
                     _ => false,
                 };
                 if is_event {
-                    // TODO: Currently we are not doing anything with events. Perhaps they should be dumped onto a stream? Perhaps many streams each representing a listener to an event?
+                    let obj = match &frame {
+                        serde_json::Value::Object(obj) => obj,
+                        _ => {
+                            return Err("Got back a json value that wasn't an object.".into());
+                        }
+                    };
+                    match obj.get("event") {
+                        Some(orig @ serde_json::Value::String(s)) => {
+                            println!("Notifying listeners for event {}", s);
+                            let listeners = {
+                                let db_handle = event_listeners.lock().unwrap();
+                                match (*db_handle).get(s) {
+                                    None => Vec::new(),
+                                    Some(listeners) => listeners.to_owned(),
+                                }
+                            };
+                            for listener in listeners {
+                                listener.send(frame.clone()).await;
+                            }
+                        }
+                        _ => return Err("Event with no string value".into()),
+                    };
                     return Ok(());
                 }
                 // get the request id and push the result into the channel
@@ -99,10 +129,6 @@ impl MpvCtl {
                             return Err("Got back a json value that wasn't an object.".into());
                         }
                     };
-                    match obj.get("event") {
-                        Some(event) => {}
-                        _ => (),
-                    };
                     match obj.get("request_id") {
                         Some(serde_json::Value::Number(reqid)) if reqid.is_u64() => {
                             reqid.as_u64().unwrap()