refactor: Use InstanceHandle for Boblight server

Author: alixinne

src/api/boblight.rs

@@ -1,11 +1,11 @@
 use std::sync::Arc;
 
 use thiserror::Error;
-use tokio::sync::mpsc::Sender;
 
 use crate::{
     global::{InputMessage, InputMessageData, InputSourceHandle, Message},
-    models::{Color, InstanceConfig},
+    instance::{InstanceHandle, InstanceHandleError},
+    models::Color,
 };
 
 pub mod message;
@@ -17,27 +17,27 @@ pub enum BoblightApiError {
     Broadcast(#[from] tokio::sync::mpsc::error::SendError<InputMessage>),
     #[error("missing command data in protobuf frame")]
     MissingCommand,
+    #[error("invalid instance")]
+    InvalidInstance(#[from] InstanceHandleError),
 }
 
 pub struct ClientConnection {
     handle: InputSourceHandle<InputMessage>,
     led_colors: Vec<Color>,
     priority: i32,
-    tx: Sender<InputMessage>,
-    instance: Arc<InstanceConfig>,
+    instance: InstanceHandle,
 }
 
 impl ClientConnection {
     pub fn new(
         handle: InputSourceHandle<InputMessage>,
-        tx: Sender<InputMessage>,
-        instance: Arc<InstanceConfig>,
+        led_count: usize,
+        instance: InstanceHandle,
     ) -> Self {
         Self {
             handle,
-            led_colors: vec![Color::default(); instance.leds.leds.len()],
+            led_colors: vec![Color::default(); led_count],
             priority: 128,
-            tx,
             instance,
         }
     }
@@ -51,8 +51,9 @@ impl ClientConnection {
         }
     }
 
-    async fn sync(&self) -> Result<(), tokio::sync::mpsc::error::SendError<InputMessage>> {
-        self.tx
+    async fn sync(&self) -> Result<(), BoblightApiError> {
+        Ok(self
+            .instance
             .send(InputMessage::new(
                 self.handle.id(),
                 crate::component::ComponentName::BoblightServer,
@@ -62,7 +63,7 @@ impl ClientConnection {
                     led_colors: Arc::new(self.led_colors.clone()),
                 },
             ))
-            .await
+            .await?)
     }
 
     pub async fn handle_request(
@@ -75,7 +76,7 @@ impl ClientConnection {
             BoblightRequest::Get(get) => match get {
                 message::GetArg::Version => Ok(Some(BoblightResponse::Version)),
                 message::GetArg::Lights => Ok(Some(BoblightResponse::Lights {
-                    leds: self.instance.leds.leds.clone(),
+                    leds: self.instance.config().await?.leds.leds.clone(),
                 })),
             },
             BoblightRequest::Set(set) => {

src/instance.rs

@@ -70,23 +70,22 @@ impl Instance {
         let muxer = PriorityMuxer::new(global.clone()).await;
         let core = Core::new(&config).await;
 
+        let (tx, handle_rx) = mpsc::channel(1);
+        let id = config.instance.id;
+        let handle = InstanceHandle { id, tx, local_tx };
+
         let config = Arc::new(config);
         let _boblight_server = if config.boblight_server.enable {
             let server_handle = servers::bind(
                 "Boblight",
                 config.boblight_server.clone(),
                 global.clone(),
                 {
-                    let instance = config.clone();
-                    let local_tx = local_tx.clone();
+                    let led_count = config.leds.leds.len();
+                    let handle = handle.clone();
 
                     move |tcp, global| {
-                        servers::boblight::handle_client(
-                            tcp,
-                            local_tx.clone(),
-                            instance.clone(),
-                            global,
-                        )
+                        servers::boblight::handle_client(tcp, led_count, handle.clone(), global)
                     }
                 },
             )
@@ -104,9 +103,6 @@ impl Instance {
             None
         };
 
-        let (tx, handle_rx) = mpsc::channel(1);
-        let id = config.instance.id;
-
         (
             Self {
                 config,
@@ -118,7 +114,7 @@ impl Instance {
                 core,
                 _boblight_server,
             },
-            InstanceHandle { id, tx, local_tx },
+            handle,
         )
     }
 
@@ -261,8 +257,8 @@ pub enum InstanceHandleError {
     Dropped,
 }
 
-impl From<tokio::sync::mpsc::error::SendError<InstanceMessage>> for InstanceHandleError {
-    fn from(_: tokio::sync::mpsc::error::SendError<InstanceMessage>) -> Self {
+impl<T> From<tokio::sync::mpsc::error::SendError<T>> for InstanceHandleError {
+    fn from(_: tokio::sync::mpsc::error::SendError<T>) -> Self {
         Self::Dropped
     }
 }
@@ -278,6 +274,10 @@ impl InstanceHandle {
         self.id
     }
 
+    pub async fn send(&self, input: InputMessage) -> Result<(), InstanceHandleError> {
+        Ok(self.local_tx.send(input).await?)
+    }
+
     pub async fn current_priorities(&self) -> Result<Vec<PriorityInfo>, InstanceHandleError> {
         let (tx, rx) = oneshot::channel();
         self.tx.send(InstanceMessage::PriorityInfo(tx)).await?;

src/servers/boblight.rs

@@ -1,17 +1,16 @@
 //! Boblight protocol server implementation
 
 use std::net::SocketAddr;
-use std::sync::Arc;
 
 use futures::prelude::*;
 use thiserror::Error;
-use tokio::{net::TcpStream, sync::mpsc::Sender};
+use tokio::net::TcpStream;
 use tokio_util::codec::Framed;
 
 use crate::{
     api::boblight::{self, BoblightApiError},
-    global::{Global, InputMessage, InputSourceName},
-    models::InstanceConfig,
+    global::{Global, InputSourceName},
+    instance::InstanceHandle,
 };
 
 /// Boblight protocol codec definition
@@ -30,8 +29,8 @@ pub enum BoblightServerError {
 
 pub async fn handle_client(
     (socket, peer_addr): (TcpStream, SocketAddr),
-    tx: Sender<InputMessage>,
-    instance: Arc<InstanceConfig>,
+    led_count: usize,
+    instance: InstanceHandle,
     global: Global,
 ) -> Result<(), BoblightServerError> {
     debug!("accepted new connection from {}", peer_addr,);
@@ -43,7 +42,8 @@ pub async fn handle_client(
         .register_input_source(InputSourceName::Boblight { peer_addr }, None)
         .await
         .unwrap();
-    let mut connection = boblight::ClientConnection::new(source_handle, tx, instance);
+
+    let mut connection = boblight::ClientConnection::new(source_handle, led_count, instance);
 
     while let Some(request) = reader.next().await {
         trace!("({}) processing request: {:?}", peer_addr, request);