refactor: Add InstanceHandle type

Author: alixinne

src/global.rs

@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 use std::net::SocketAddr;
 use std::sync::Arc;
 
@@ -12,7 +12,7 @@ pub use input_message::*;
 mod input_source;
 pub use input_source::*;
 
-use crate::{component::ComponentName, models::Config};
+use crate::{component::ComponentName, instance::InstanceHandle, models::Config};
 
 pub trait Message: Sized {
     type Data;
@@ -75,6 +75,14 @@ impl Global {
         self.0.read().await.input_tx.subscribe()
     }
 
+    pub async fn register_instance(&self, handle: InstanceHandle) {
+        self.0.write().await.register_instance(handle);
+    }
+
+    pub async fn get_instance(&self, id: i32) -> Option<InstanceHandle> {
+        self.0.read().await.instances.get(&id).cloned()
+    }
+
     pub async fn read_config<T>(&self, f: impl FnOnce(&Config) -> T) -> T {
         let data = self.0.read().await;
         f(&data.config)
@@ -94,6 +102,7 @@ pub struct GlobalData {
     input_sources: HashMap<usize, Arc<InputSource<InputMessage>>>,
     next_input_source_id: usize,
     config: Config,
+    instances: BTreeMap<i32, InstanceHandle>,
 }
 
 impl GlobalData {
@@ -105,6 +114,7 @@ impl GlobalData {
             input_sources: Default::default(),
             next_input_source_id: 1,
             config: config.clone(),
+            instances: Default::default(),
         }
     }
 
@@ -134,4 +144,8 @@ impl GlobalData {
             info!("unregistered input source {}", *is);
         }
     }
+
+    fn register_instance(&mut self, handle: InstanceHandle) {
+        self.instances.insert(handle.id(), handle);
+    }
 }

src/instance.rs

@@ -1,9 +1,12 @@
 use std::sync::Arc;
 
 use thiserror::Error;
-use tokio::sync::broadcast;
-use tokio::{select, sync::mpsc};
+use tokio::{
+    select,
+    sync::{broadcast, mpsc, oneshot},
+};
 
+use crate::api::types::PriorityInfo;
 use crate::models::Color;
 use crate::{
     global::{Global, InputMessage},
@@ -37,7 +40,9 @@ pub enum InstanceError {
 }
 
 pub struct Instance {
+    config: Arc<InstanceConfig>,
     device: InstanceDevice,
+    handle_rx: mpsc::Receiver<InstanceMessage>,
     receiver: broadcast::Receiver<InputMessage>,
     local_receiver: mpsc::Receiver<InputMessage>,
     muxer: PriorityMuxer,
@@ -46,7 +51,7 @@ pub struct Instance {
 }
 
 impl Instance {
-    pub async fn new(global: Global, config: InstanceConfig) -> Self {
+    pub async fn new(global: Global, config: InstanceConfig) -> (Self, InstanceHandle) {
         let device: InstanceDevice =
             Device::new(&config.instance.friendly_name, config.device.clone())
                 .await
@@ -92,14 +97,22 @@ impl Instance {
             None
         };
 
-        Self {
-            device,
-            receiver,
-            local_receiver,
-            muxer,
-            core,
-            _boblight_server,
-        }
+        let (tx, handle_rx) = mpsc::channel(1);
+        let id = config.instance.id;
+
+        (
+            Self {
+                config,
+                device,
+                handle_rx,
+                receiver,
+                local_receiver,
+                muxer,
+                core,
+                _boblight_server,
+            },
+            InstanceHandle { id, tx },
+        )
     }
 
     async fn on_input_message(&mut self, message: InputMessage) {
@@ -109,6 +122,19 @@ impl Instance {
         }
     }
 
+    pub fn id(&self) -> i32 {
+        self.config.instance.id
+    }
+
+    async fn handle_instance_message(&mut self, message: InstanceMessage) {
+        match message {
+            InstanceMessage::PriorityInfo(tx) => {
+                // unwrap: the receiver should not have dropped
+                tx.send(self.muxer.current_priorities().await).unwrap();
+            }
+        }
+    }
+
     pub async fn run(mut self) -> Result<(), InstanceError> {
         loop {
             select! {
@@ -123,7 +149,7 @@ impl Instance {
                         },
                         Err(tokio::sync::broadcast::error::RecvError::Closed) => {
                             // No more input messages
-                            return Ok(());
+                            break Ok(());
                         },
                         Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
                             warn!("skipped {} input messages", skipped);
@@ -147,6 +173,14 @@ impl Instance {
                     // LED data changed
                     self.device.set_led_data(led_data).await?;
                 },
+                message = self.handle_rx.recv() => {
+                    if let Some(message) = message {
+                        self.handle_instance_message(message).await;
+                    } else {
+                        // If the handle is dropped, it means the instance was unregistered
+                        break Ok(());
+                    }
+                }
             }
         }
     }
@@ -181,3 +215,32 @@ impl From<Result<Device, DeviceError>> for InstanceDevice {
         Self { inner }
     }
 }
+
+#[derive(Debug)]
+enum InstanceMessage {
+    PriorityInfo(oneshot::Sender<Vec<PriorityInfo>>),
+}
+
+#[derive(Clone)]
+pub struct InstanceHandle {
+    id: i32,
+    tx: mpsc::Sender<InstanceMessage>,
+}
+
+impl InstanceHandle {
+    pub fn id(&self) -> i32 {
+        self.id
+    }
+
+    pub async fn current_priorities(&self) -> Vec<PriorityInfo> {
+        let (tx, rx) = oneshot::channel();
+
+        // TODO: Don't unwrap and propagate?
+        self.tx
+            .send(InstanceMessage::PriorityInfo(tx))
+            .await
+            .unwrap();
+        // unwrap: if the previous didn't fail, the instance will be there to answer
+        rx.await.unwrap()
+    }
+}

src/main.rs

@@ -33,7 +33,11 @@ async fn run(opts: Opts) -> color_eyre::eyre::Result<()> {
 
     // Initialize and spawn the devices
     for (_, inst) in &config.instances {
-        let inst = hyperion::instance::Instance::new(global.clone(), inst.clone()).await;
+        // Create the instance
+        let (inst, handle) = hyperion::instance::Instance::new(global.clone(), inst.clone()).await;
+        // Register the instance globally using its handle
+        global.register_instance(handle).await;
+        // Run the instance futures
         tokio::spawn(async move {
             let result = inst.run().await;