src-tauri/src/display/manager.rs

@@ -5,10 +5,7 @@ use std::{
 
 use ddc_hi::Display;
 use paris::{error, info, warn};
-use tokio::{
+use tokio::{sync::{watch, OnceCell, RwLock}, task::yield_now};
-    sync::{broadcast, watch, OnceCell, RwLock},
-    task::yield_now,
-};
 
 use crate::rpc::{BoardMessageChannels, DisplaySetting};
 
@@ -78,22 +75,7 @@ impl DisplayManager {
             let channels = BoardMessageChannels::global().await;
             let mut request_rx = channels.display_setting_request_sender.subscribe();
 
-            loop {
+            while let Ok(message) = request_rx.recv().await {
-                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);
@@ -102,6 +84,7 @@ impl DisplayManager {
                     continue;
                 }
 
+
                 let display = display.unwrap().write().await;
                 let result = match message.setting {
                     DisplaySetting::Brightness(value) => display.set_brightness(value as u16).await,
@@ -139,8 +122,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,16 +3,15 @@ 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, volume::VolumeManager};
+use crate::rpc::DisplaySettingRequest;
 
-use super::{BoardConnectStatus, BoardInfo, BoardMessageChannels};
+use super::{BoardConnectStatus, BoardInfo};
 
 #[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 {
@@ -21,21 +20,25 @@ 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.clone();
+        let info = self.info.read().await;
-        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;
 
@@ -79,52 +82,9 @@ 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 {
@@ -195,14 +155,12 @@ impl Board {
 
 impl Drop for Board {
     fn drop(&mut self) {
-        info!("board drop");
-
         if let Some(handler) = self.listen_handler.take() {
-            handler.abort();
+            info!("aborting listen handler");
-        }
+            tokio::task::block_in_place(move || {
-
+                handler.abort();
-        if let Some(handler) = self.volume_changed_subscriber_handler.take() {
+            });
-            handler.abort();
+            info!("listen handler aborted");
         }
     }
 }

src-tauri/src/rpc/channels.rs

@@ -7,7 +7,6 @@ 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 {
@@ -24,13 +23,9 @@ 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,22 +1,23 @@
-use std::{mem, sync::Arc};
+use std::{
+    mem,
+    sync::{Arc, RwLock},
+};
 
 use coreaudio::{
     audio_unit::macos_helpers::get_default_device_id,
     sys::{
         kAudioHardwareServiceDeviceProperty_VirtualMasterVolume, kAudioObjectPropertyScopeOutput,
-        AudioObjectGetPropertyData, AudioObjectHasProperty, AudioObjectPropertyAddress,
+        AudioObjectHasProperty, AudioObjectPropertyAddress, AudioObjectSetPropertyData,
-        AudioObjectSetPropertyData,
     },
 };
 use paris::error;
-use tokio::sync::{OnceCell, RwLock};
+use tokio::sync::OnceCell;
 
 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 {
@@ -32,11 +33,9 @@ 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
     }
@@ -56,36 +55,6 @@ 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);
 
@@ -129,75 +98,4 @@ 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();
-            });
-        }
-    }
 }