feat: 当新板子上线或音量变化时，推送当前音量给板子。

src-tauri/src/display/manager.rs

@@ -5,7 +5,10 @@ use std::{
 
 use ddc_hi::Display;
 use paris::{error, info, warn};
-use tokio::{sync::{watch, OnceCell, RwLock}, task::yield_now};
+use tokio::{
+    sync::{broadcast, watch, OnceCell, RwLock},
+    task::yield_now,
+};
 
 use crate::rpc::{BoardMessageChannels, DisplaySetting};
 
@@ -75,7 +78,22 @@ impl DisplayManager {
             let channels = BoardMessageChannels::global().await;
             let mut request_rx = channels.display_setting_request_sender.subscribe();
 
-            while let Ok(message) = request_rx.recv().await {
+            loop {
+                if let Err(err) = request_rx.recv().await {
+                    match err {
+                        broadcast::error::RecvError::Closed => {
+                            info!("display setting request channel closed");
+                            break;
+                        }
+                        broadcast::error::RecvError::Lagged(_) => {
+                            warn!("display setting request channel lagged");
+                            continue;
+                        }
+                    }
+                }
+
+                let message = request_rx.recv().await.unwrap();
+
                 let displays = displays.write().await;
 
                 let display = displays.get(message.display_index);
@@ -84,7 +102,6 @@ impl DisplayManager {
                     continue;
                 }
 
-
                 let display = display.unwrap().write().await;
                 let result = match message.setting {
                     DisplaySetting::Brightness(value) => display.set_brightness(value as u16).await,
@@ -122,8 +139,8 @@ impl DisplayManager {
 
 impl Drop for DisplayManager {
     fn drop(&mut self) {
+        log::info!("dropping display manager=============");
         if let Some(handler) = self.setting_request_handler.take() {
-            info!("abort display setting request handler");
             handler.abort();
         }
     }

src-tauri/src/rpc/board.rs

@@ -3,15 +3,16 @@ use std::{sync::Arc, time::Duration};
 use paris::{error, info, warn};
 use tokio::{io, net::UdpSocket, sync::RwLock, task::yield_now, time::timeout};
 
-use crate::rpc::DisplaySettingRequest;
+use crate::{rpc::DisplaySettingRequest, volume::VolumeManager};
 
-use super::{BoardConnectStatus, BoardInfo};
+use super::{BoardConnectStatus, BoardInfo, BoardMessageChannels};
 
 #[derive(Debug)]
 pub struct Board {
     pub info: Arc<RwLock<BoardInfo>>,
     socket: Option<Arc<UdpSocket>>,
     listen_handler: Option<tokio::task::JoinHandle<()>>,
+    volume_changed_subscriber_handler: Option<tokio::task::JoinHandle<()>>,
 }
 
 impl Board {
@@ -20,25 +21,21 @@ impl Board {
             info: Arc::new(RwLock::new(info)),
             socket: None,
             listen_handler: None,
+            volume_changed_subscriber_handler: None,
         }
     }
 
     pub async fn init_socket(&mut self) -> anyhow::Result<()> {
-        let info = self.info.read().await;
+        let info = self.info.clone();
+        let info = info.read().await;
         let socket = UdpSocket::bind("0.0.0.0:0").await?;
 
         socket.connect((info.address, info.port)).await?;
         let socket = Arc::new(socket);
         self.socket = Some(socket.clone());
 
-        let info = self.info.clone();
-
         let handler = tokio::spawn(async move {
             let mut buf = [0u8; 128];
-            if let Err(err) = socket.readable().await {
-                error!("socket read error: {:?}", err);
-                return;
-            }
 
             let board_message_channels = crate::rpc::channels::BoardMessageChannels::global().await;
 
@@ -82,9 +79,52 @@ impl Board {
         });
         self.listen_handler = Some(handler);
 
+        self.subscribe_volume_changed().await;
+
         Ok(())
     }
 
+    async fn subscribe_volume_changed(&mut self) {
+        let channel = BoardMessageChannels::global().await;
+        let mut volume_changed_rx = channel.volume_changed_sender.subscribe();
+        let info = self.info.clone();
+        let socket = self.socket.clone();
+
+        let handler = tokio::spawn(async move {
+            while let Ok(volume) = volume_changed_rx.recv().await {
+                let info = info.read().await;
+                if socket.is_none() || info.connect_status != BoardConnectStatus::Connected {
+                    log::info!("board is not connected, skip send volume changed");
+                    continue;
+                }
+
+                let socket = socket.as_ref().unwrap();
+
+                let mut buf = [0u8; 2];
+                buf[0] = 4;
+                buf[1] = (volume * 100.0) as u8;
+
+                if let Err(err) = socket.send(&buf).await {
+                    log::warn!("send volume changed failed: {:?}", err);
+                }
+            }
+        });
+
+        let volume_manager = VolumeManager::global().await;
+        let volume = volume_manager.get_volume().await;
+
+        if let Some(socket) = self.socket.as_ref() {
+            let buf = [4, (volume * 100.0) as u8];
+            if let Err(err) = socket.send(&buf).await {
+                log::warn!("send volume failed: {:?}", err);
+            }
+        } else {
+            log::warn!("socket is none, skip send volume");
+        }
+
+        self.volume_changed_subscriber_handler = Some(handler);
+    }
+
     pub async fn send_colors(&self, buf: &[u8]) {
         let info = self.info.read().await;
         if self.socket.is_none() || info.connect_status != BoardConnectStatus::Connected {
@@ -155,12 +195,14 @@ impl Board {
 
 impl Drop for Board {
     fn drop(&mut self) {
+        info!("board drop");
+
         if let Some(handler) = self.listen_handler.take() {
-            info!("aborting listen handler");
-            tokio::task::block_in_place(move || {
             handler.abort();
-            });
+        }
-            info!("listen handler aborted");
+
+        if let Some(handler) = self.volume_changed_subscriber_handler.take() {
+            handler.abort();
         }
     }
 }

src-tauri/src/rpc/channels.rs

@@ -7,6 +7,7 @@ use super::DisplaySettingRequest;
 pub struct BoardMessageChannels {
     pub display_setting_request_sender: Arc<broadcast::Sender<DisplaySettingRequest>>,
     pub volume_setting_request_sender: Arc<broadcast::Sender<f32>>,
+    pub volume_changed_sender: Arc<broadcast::Sender<f32>>,
 }
 
 impl BoardMessageChannels {
@@ -23,9 +24,13 @@ impl BoardMessageChannels {
         let (volume_setting_request_sender, _) = broadcast::channel(16);
         let volume_setting_request_sender = Arc::new(volume_setting_request_sender);
 
+        let (volume_changed_sender, _) = broadcast::channel(2);
+        let volume_changed_sender = Arc::new(volume_changed_sender);
+
         Self {
             display_setting_request_sender,
             volume_setting_request_sender,
+            volume_changed_sender,
         }
     }
 }

src-tauri/src/volume/manager.rs

@@ -1,23 +1,22 @@
-use std::{
+use std::{mem, sync::Arc};
-    mem,
-    sync::{Arc, RwLock},
-};
 
 use coreaudio::{
     audio_unit::macos_helpers::get_default_device_id,
     sys::{
         kAudioHardwareServiceDeviceProperty_VirtualMasterVolume, kAudioObjectPropertyScopeOutput,
-        AudioObjectHasProperty, AudioObjectPropertyAddress, AudioObjectSetPropertyData,
+        AudioObjectGetPropertyData, AudioObjectHasProperty, AudioObjectPropertyAddress,
+        AudioObjectSetPropertyData,
     },
 };
 use paris::error;
-use tokio::sync::OnceCell;
+use tokio::sync::{OnceCell, RwLock};
 
 use crate::rpc::BoardMessageChannels;
 
 pub struct VolumeManager {
     current_volume: Arc<RwLock<f32>>,
     handler: Option<tokio::task::JoinHandle<()>>,
+    read_handler: Option<tokio::task::JoinHandle<()>>,
 }
 
 impl VolumeManager {
@@ -33,9 +32,11 @@ impl VolumeManager {
         let mut instance = Self {
             current_volume: Arc::new(RwLock::new(0.0)),
             handler: None,
+            read_handler: None,
         };
 
         instance.subscribe_volume_setting_request();
+        instance.auto_read_volume();
 
         instance
     }
@@ -55,6 +56,36 @@ impl VolumeManager {
         self.handler = Some(handler);
     }
 
+    fn auto_read_volume(&mut self) {
+        let current_volume = self.current_volume.clone();
+
+        let handler = tokio::spawn(async move {
+            let channel = BoardMessageChannels::global().await;
+            let volume_changed_tx = channel.volume_changed_sender.clone();
+            loop {
+                match Self::read_volume() {
+                    Ok(value) => {
+                        let mut volume = current_volume.write().await;
+                        if *volume != value {
+                            if let Err(err) = volume_changed_tx.send(value) {
+                                error!("failed to send volume changed event: {}", err);
+                            }
+                        }
+
+                        *volume = value;
+                    }
+                    Err(err) => {
+                        error!("failed to read volume: {}", err);
+                    }
+                }
+
+                tokio::time::sleep(std::time::Duration::from_secs(10)).await;
+            }
+        });
+
+        self.read_handler = Some(handler);
+    }
+
     fn set_volume(volume: f32) -> anyhow::Result<()> {
         log::debug!("set volume: {}", volume);
 
@@ -98,4 +129,75 @@ impl VolumeManager {
 
         Ok(())
     }
+
+    fn read_volume() -> anyhow::Result<f32> {
+        let device_id = get_default_device_id(false);
+
+        if device_id.is_none() {
+            anyhow::bail!("default audio output device is not found.");
+        }
+
+        let device_id = device_id.unwrap();
+
+        let address = AudioObjectPropertyAddress {
+            mSelector: kAudioHardwareServiceDeviceProperty_VirtualMasterVolume,
+            mScope: kAudioObjectPropertyScopeOutput,
+            mElement: 0,
+        };
+
+        log::debug!("device id: {}", device_id);
+        log::debug!("address: {:?}", address);
+
+        if 0 == unsafe { AudioObjectHasProperty(device_id, &address) } {
+            anyhow::bail!("Can not get audio property");
+        }
+
+        let mut size = mem::size_of::<f32>() as u32;
+
+        let mut volume = 0.0f32;
+
+        let result = unsafe {
+            AudioObjectGetPropertyData(
+                device_id,
+                &address,
+                0,
+                std::ptr::null(),
+                &mut size,
+                &mut volume as *mut f32 as *mut std::ffi::c_void,
+            )
+        };
+
+        if result != 0 {
+            anyhow::bail!("Can not get audio property. result: {}", result);
+        }
+
+        if size != mem::size_of::<f32>() as u32 {
+            anyhow::bail!("Can not get audio property. data size is not matched.");
+        }
+
+        log::debug!("current system volume of primary device: {}", volume);
+
+        Ok(volume)
+    }
+
+    pub async fn get_volume(&self) -> f32 {
+        self.current_volume.read().await.clone()
+    }
+}
+
+impl Drop for VolumeManager {
+    fn drop(&mut self) {
+        log::info!("drop volume manager");
+        if let Some(handler) = self.handler.take() {
+            tokio::task::block_in_place(move || {
+                handler.abort();
+            });
+        }
+
+        if let Some(handler) = self.read_handler.take() {
+            tokio::task::block_in_place(move || {
+                handler.abort();
+            });
+        }
+    }
 }