chore: 更新。

package.json

@@ -10,13 +10,13 @@
     "tauri": "tauri"
   },
   "dependencies": {
-    "@emotion/react": "^11.10.5",
+    "@emotion/react": "^11.10.6",
-    "@emotion/styled": "^11.10.5",
+    "@emotion/styled": "^11.10.6",
-    "@fortawesome/fontawesome-svg-core": "^6.2.1",
+    "@fortawesome/fontawesome-svg-core": "^6.3.0",
-    "@fortawesome/free-regular-svg-icons": "^6.2.1",
+    "@fortawesome/free-regular-svg-icons": "^6.3.0",
-    "@fortawesome/free-solid-svg-icons": "^6.2.1",
+    "@fortawesome/free-solid-svg-icons": "^6.3.0",
     "@fortawesome/react-fontawesome": "^0.2.0",
-    "@mui/material": "^5.11.4",
+    "@mui/material": "^5.11.13",
     "@tauri-apps/api": "^1.2.0",
     "clsx": "^1.2.1",
     "debug": "^4.3.4",
@@ -27,28 +27,28 @@
     "react-dom": "^18.2.0"
   },
   "devDependencies": {
-    "@babel/plugin-transform-react-jsx": "^7.20.7",
+    "@babel/plugin-transform-react-jsx": "^7.21.0",
     "@emotion/babel-plugin-jsx-pragmatic": "^0.2.0",
     "@emotion/serialize": "^1.1.1",
-    "@tauri-apps/cli": "^1.2.2",
+    "@tauri-apps/cli": "^1.2.3",
     "@types/debug": "^4.1.7",
-    "@types/node": "^18.11.18",
+    "@types/node": "^18.15.3",
-    "@types/ramda": "^0.28.20",
+    "@types/ramda": "^0.28.23",
-    "@types/react": "^18.0.26",
+    "@types/react": "^18.0.28",
-    "@types/react-dom": "^18.0.10",
+    "@types/react-dom": "^18.0.11",
     "@vitejs/plugin-react": "^2.2.0",
-    "autoprefixer": "^10.4.13",
+    "autoprefixer": "^10.4.14",
     "babel-plugin-macros": "^3.1.0",
-    "eslint-config-prettier": "^8.6.0",
+    "eslint-config-prettier": "^8.7.0",
-    "eslint-plugin-import": "^2.27.4",
+    "eslint-plugin-import": "^2.27.5",
     "eslint-plugin-jsx-a11y": "^6.7.1",
     "eslint-plugin-prettier": "^4.2.1",
     "eslint-plugin-simple-import-sort": "^8.0.0",
     "postcss": "^8.4.21",
-    "prettier": "^2.8.3",
+    "prettier": "^2.8.4",
-    "tailwindcss": "^3.2.4",
+    "tailwindcss": "^3.2.7",
     "twin.macro": "^3.1.0",
-    "typescript": "^4.9.4",
+    "typescript": "^4.9.5",
     "vite": "^3.2.5"
   }
 }

src-tauri/Cargo.toml

@@ -28,7 +28,6 @@ tokio = { version = "1.22.0", features = ["full"] }
 tracing = "0.1.37"
 tracing-subscriber = "0.3.16"
 hex = "0.4.3"
-rumqttc = "0.17.0"
 time = { version = "0.3.17", features = ["formatting"] }
 color_space = "0.5.3"
 futures = "0.3.25"
@@ -36,14 +35,17 @@ either = "1.8.0"
 image = "0.24.5"
 mdns = "3.0.0"
 macos-app-nap = "0.0.1"
+ddc-hi = "0.4.1"
+redb = "0.13.0"
+paho-mqtt = "0.12.0"
 
 [features]
 # by default Tauri runs in production mode
 # when `tauri dev` runs it is executed with `cargo run --no-default-features` if `devPath` is an URL
-default = [ "custom-protocol" ]
+default = ["custom-protocol"]
 # this feature is used used for production builds where `devPath` points to the filesystem
 # DO NOT remove this
-custom-protocol = [ "tauri/custom-protocol" ]
+custom-protocol = ["tauri/custom-protocol"]
 
 [dev-dependencies]
 test_dir = "0.2.0"

src-tauri/src/core/core.rs

@@ -75,7 +75,8 @@ impl CoreManager {
                     colors.push(color);
                 }
                 hue = (hue + 1.0) % 360.0;
-                match rpc::manager::Manager::global()
+                match rpc::Manager::global()
+                    .await
                     .publish_led_colors(&colors)
                     .await
                 {
@@ -139,7 +140,6 @@ impl CoreManager {
         tokio::spawn(async move {
             let mut global_sub_pixels = HashMap::new();
             while let Some(screenshot) = rx.recv().await {
-                let start_at = Instant::now();
                 let colors = screenshot.get_colors();
                 let config = screenshot.get_config();
                 for (colors, config) in vec![
@@ -183,11 +183,12 @@ impl CoreManager {
                 if global_sub_pixels.len() >= total_colors_count * 3 {
                     let mut colors = vec![];
                     for index in 0..global_sub_pixels.len() {
-                        colors.push(*global_sub_pixels.get(&index).unwrap());
+                        colors.push(*global_sub_pixels.get(&index).unwrap_or(&100));
                     }
                     // info!("{:?}", colors);
                     global_sub_pixels = HashMap::new();
-                    match rpc::manager::Manager::global()
+                    match rpc::Manager::global()
+                        .await
                         .publish_led_sub_pixels(colors)
                         .await
                     {

src-tauri/src/db/db.rs

@@ -0,0 +1,25 @@
+use std::env::current_dir;
+
+use once_cell::sync::OnceCell;
+use redb::Database;
+use tauri::api::path::config_dir;
+
+use crate::picker;
+
+trait GlobalDatabase<T> {
+    fn global() -> &'static T;
+}
+
+impl GlobalDatabase<Database> for Database {
+    fn global() -> &'static Database {
+        static GLOBAL_DATABASE: OnceCell<Database> = OnceCell::new();
+
+        GLOBAL_DATABASE.get_or_init(|| {
+            let path = config_dir()
+                .unwrap_or(current_dir().unwrap())
+                .join("main.redb");
+            let db = Database::create(path).unwrap();
+            return db;
+        })
+    }
+}

src-tauri/src/db/mod.rs

@@ -0,0 +1,3 @@
+mod db;
+
+pub use db::*;

src-tauri/src/display/brightness.rs

@@ -0,0 +1,13 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Copy, Serialize, Deserialize, Debug)]
+pub enum Brightness {
+    Relative(i16),
+    Absolute(u16),
+}
+
+#[derive(Clone, Copy, Serialize, Deserialize, Debug)]
+pub struct DisplayBrightness {
+    pub brightness: Brightness,
+    pub display_index: usize,
+}

src-tauri/src/display/display_config.rs

@@ -0,0 +1,36 @@
+use std::time::SystemTime;
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Copy, Serialize, Deserialize, Debug)]
+pub struct DisplayConfig {
+    pub id: usize,
+    pub brightness: u16,
+    pub max_brightness: u16,
+    pub min_brightness: u16,
+    pub contrast: u16,
+    pub max_contrast: u16,
+    pub min_contrast: u16,
+    pub mode: u16,
+    pub max_mode: u16,
+    pub min_mode: u16,
+    pub last_modified_at: SystemTime,
+}
+
+impl DisplayConfig {
+    pub fn default(index: usize) -> Self {
+        Self {
+            id: index,
+            brightness: 30,
+            contrast: 50,
+            mode: 0,
+            last_modified_at: SystemTime::now(),
+            max_brightness: 100,
+            min_brightness: 0,
+            max_contrast: 100,
+            min_contrast: 0,
+            max_mode: 15,
+            min_mode: 0,
+        }
+    }
+}

src-tauri/src/display/manager.rs

@@ -0,0 +1,186 @@
+use std::{
+    borrow::Borrow,
+    collections::HashMap,
+    ops::Sub,
+    sync::Arc,
+    time::{Duration, SystemTime},
+};
+
+use base64::Config;
+use ddc_hi::Display;
+use paris::{error, info};
+use tauri::async_runtime::Mutex;
+use tokio::sync::{broadcast, OwnedMutexGuard};
+use tracing::warn;
+
+use crate::{display::Brightness, models, rpc};
+
+use super::{display_config::DisplayConfig, DisplayBrightness};
+use ddc_hi::Ddc;
+
+pub struct Manager {
+    displays: Arc<Mutex<HashMap<usize, Arc<Mutex<DisplayConfig>>>>>,
+}
+
+impl Manager {
+    pub fn global() -> &'static Self {
+        static DISPLAY_MANAGER: once_cell::sync::OnceCell<Manager> =
+            once_cell::sync::OnceCell::new();
+
+        DISPLAY_MANAGER.get_or_init(|| Self::create())
+    }
+
+    pub fn create() -> Self {
+        let instance = Self {
+            displays: Arc::new(Mutex::new(HashMap::new())),
+        };
+        instance
+    }
+
+    pub async fn subscribe_display_brightness(&self) {
+        let rpc = rpc::Manager::global().await;
+
+        let mut rx = rpc.client().subscribe_change_display_brightness_rx();
+
+        loop {
+            if let Ok(display_brightness) = rx.recv().await {
+                if let Err(err) = self.set_display_brightness(display_brightness).await {
+                    error!("set_display_brightness failed. {:?}", err);
+                }
+            }
+        }
+    }
+
+    fn read_display_config_by_ddc(index: usize) -> anyhow::Result<DisplayConfig> {
+        let mut displays = Display::enumerate();
+        match displays.get_mut(index) {
+            Some(display) => {
+                let mut config = DisplayConfig::default(index);
+                match display.handle.get_vcp_feature(0x10) {
+                    Ok(value) => {
+                        config.max_brightness = value.maximum();
+                        config.min_brightness = 0;
+                        config.brightness = value.value();
+                    }
+                    Err(_) => {}
+                };
+                match display.handle.get_vcp_feature(0x12) {
+                    Ok(value) => {
+                        config.max_contrast = value.maximum();
+                        config.min_contrast = 0;
+                        config.contrast = value.value();
+                    }
+                    Err(_) => {}
+                };
+                match display.handle.get_vcp_feature(0xdc) {
+                    Ok(value) => {
+                        config.max_mode = value.maximum();
+                        config.min_mode = 0;
+                        config.mode = value.value();
+                    }
+                    Err(_) => {}
+                };
+
+                Ok(config)
+            }
+            None => anyhow::bail!("display#{} is missed.", index),
+        }
+    }
+
+    async fn get_display(&self, index: usize) -> anyhow::Result<OwnedMutexGuard<DisplayConfig>> {
+        let mut displays = self.displays.lock().await;
+        match displays.get_mut(&index) {
+            Some(config) => {
+                let mut config = config.to_owned().lock_owned().await;
+                if config.last_modified_at > SystemTime::now().sub(Duration::from_secs(10)) {
+                    info!("cached");
+                    return Ok(config);
+                }
+                return match Self::read_display_config_by_ddc(index) {
+                    Ok(config) => {
+                        let id = config.id;
+                        let value = Arc::new(Mutex::new(config));
+                        let valueGuard = value.clone().lock_owned().await;
+                        displays.insert(id, value);
+                        info!("read form ddc");
+                        Ok(valueGuard)
+                    }
+                    Err(err) => {
+                        warn!(
+                            "can not read config from display by ddc, use CACHED value. {:?}",
+                            err
+                        );
+                        config.last_modified_at = SystemTime::now();
+                        Ok(config)
+                    }
+                };
+            }
+            None => {
+                let config = Self::read_display_config_by_ddc(index).map_err(|err| {
+                    anyhow::anyhow!(
+                        "can not read config from display by ddc,use DEFAULT value. {:?}",
+                        err
+                    )
+                })?;
+                let id = config.id;
+                let value = Arc::new(Mutex::new(config));
+                let valueGuard = value.clone().lock_owned().await;
+                displays.insert(id, value);
+                Ok(valueGuard)
+            }
+        }
+    }
+
+    pub async fn set_display_brightness(
+        &self,
+        display_brightness: DisplayBrightness,
+    ) -> anyhow::Result<()> {
+        match Display::enumerate().get_mut(display_brightness.display_index) {
+            Some(display) => {
+                match self.get_display(display_brightness.display_index).await {
+                    Ok(mut config) => {
+                        let curr = config.brightness;
+                        info!("curr_brightness: {:?}", curr);
+                        let mut target = match display_brightness.brightness {
+                            Brightness::Relative(v) => curr.wrapping_add_signed(v),
+                            Brightness::Absolute(v) => v,
+                        };
+                        if target.gt(&config.max_brightness) {
+                            target = config.max_brightness;
+                        } else if target.lt(&config.min_brightness) {
+                            target = config.min_brightness;
+                        }
+                        config.brightness = target;
+                        display
+                            .handle
+                            .set_vcp_feature(0x10, target as u16)
+                            .map_err(|err| anyhow::anyhow!("can not set brightness. {:?}", err))?;
+
+                        let rpc = rpc::Manager::global().await;
+
+                        rpc.publish_desktop_cmd(
+                            format!("display{}/brightness", display_brightness.display_index).as_str(),
+                            target.to_be_bytes().to_vec(),
+                        )
+                        .await;
+                    }
+                    Err(err) => {
+                        info!(
+                            "can not get display#{} brightness. {:?}",
+                            display_brightness.display_index, err
+                        );
+                        if let Brightness::Absolute(v) = display_brightness.brightness {
+                            display.handle.set_vcp_feature(0x10, v).map_err(|err| {
+                                anyhow::anyhow!("can not set brightness. {:?}", err)
+                            })?;
+                        };
+                    }
+                };
+            }
+            None => {
+                warn!("display#{} is not found.", display_brightness.display_index);
+            }
+        }
+        Ok(())
+    }
+}

src-tauri/src/display/mod.rs

@@ -0,0 +1,9 @@
+mod brightness;
+mod manager;
+mod display_config;
+
+pub use brightness::*;
+pub use manager::*;
+
+
+

src-tauri/src/main.rs

@@ -5,24 +5,30 @@
 #![feature(bool_to_option)]
 
 mod core;
+mod db;
+mod display;
 mod picker;
 mod rpc;
+mod models;
 
 use crate::core::AmbientLightMode;
 use crate::core::CoreManager;
+use once_cell::sync::OnceCell;
 use paris::*;
 use picker::config::DisplayConfig;
 use picker::manager::Picker;
 use picker::screenshot::ScreenshotDto;
 use tauri::async_runtime::Mutex;
-use once_cell::sync::OnceCell;
 
 static GET_SCREENSHOT_LOCK: OnceCell<Mutex<bool>> = OnceCell::new();
 
 #[tauri::command]
 async fn take_snapshot() -> Vec<ScreenshotDto> {
     info!("Hi?");
-    let _lock = GET_SCREENSHOT_LOCK.get_or_init(|| Mutex::new(false)).lock().await;
+    let _lock = GET_SCREENSHOT_LOCK
+        .get_or_init(|| Mutex::new(false))
+        .lock()
+        .await;
     info!("Hi!");
     let manager = Picker::global().await;
 
@@ -89,7 +95,11 @@ async fn play_mode(target_mode: AmbientLightMode) {
 
 #[tokio::main]
 async fn main() {
-    rpc::manager::Manager::global();
+    let display_manager = display::Manager::global();
+    tokio::spawn(display_manager.subscribe_display_brightness());
+    let rpc_manager = rpc::Manager::global().await;
+    tokio::spawn(rpc_manager.listen());
+
     tauri::Builder::default()
         .invoke_handler(tauri::generate_handler![
             take_snapshot,

src-tauri/src/models/cmd_resp_with_range.rs

@@ -0,0 +1,8 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Copy, Serialize, Deserialize, Debug)]
+pub struct CmdRespWithRange<T = u16> {
+    pub value: T,
+    pub max: T,
+    pub min: T,
+}

src-tauri/src/models/config_display_cmd.rs

@@ -0,0 +1,10 @@
+use serde::{Serialize, Deserialize};
+
+use super::control_value::ControlValue;
+
+
+#[derive(Clone, Copy, Serialize, Deserialize, Debug)]
+pub struct ConfigDisplayCmd<T = ControlValue> {
+    pub display_index: usize,
+    pub value: T,
+}

src-tauri/src/models/control_value.rs

@@ -0,0 +1,8 @@
+use serde::{Serialize, Deserialize};
+
+
+#[derive(Clone, Copy, Serialize, Deserialize, Debug)]
+pub enum ControlValue<AT = u16, RT = i16> {
+    Absolute(AT),
+    Relative(RT),
+}

src-tauri/src/models/mod.rs

@@ -0,0 +1,10 @@
+mod control_value;
+mod mq_message;
+mod config_display_cmd;
+mod cmd_resp_with_range;
+
+pub use control_value::*;
+pub use mq_message::*;
+pub use config_display_cmd::*;
+pub use cmd_resp_with_range::*;
+

src-tauri/src/models/mq_message.rs

@@ -0,0 +1,17 @@
+use serde::{Serialize, Deserialize};
+
+use super::{ConfigDisplayCmd, CmdRespWithRange};
+
+
+#[derive(Clone, Copy, Serialize, Deserialize, Debug)]
+pub enum CmdMqMessage {
+    Brightness(ConfigDisplayCmd),
+    Contrast(ConfigDisplayCmd),
+    PresetMode(ConfigDisplayCmd),
+}
+#[derive(Clone, Copy, Serialize, Deserialize, Debug)]
+pub enum CmdRespMqMessage {
+    Brightness(ConfigDisplayCmd<CmdRespWithRange>),
+    Contrast(ConfigDisplayCmd<CmdRespWithRange>),
+    PresetMode(ConfigDisplayCmd<CmdRespWithRange>),
+}

src-tauri/src/picker/config/manger.rs

@@ -115,7 +115,6 @@ mod tests {
     async fn write_config_to_disk_should_be_successful() {
         let temp = TestDir::temp().create("config_dir", test_dir::FileType::Dir);
         let config_file_path = temp.path("config_dir").join("picker.config.json");
-        let manager = crate::picker::config::manger::Manager::default();
         crate::picker::config::manger::Manager::write_config_to_disk(
             config_file_path.clone(),
             &Configuration::default(),

src-tauri/src/picker/display_picker.rs

@@ -9,10 +9,6 @@ pub struct DisplayPicker {
 }
 
 impl DisplayPicker {
-    pub fn new(screen: Screen, config: DisplayConfig) -> Self {
-        Self { screen, config }
-    }
-
     pub fn from_config(config: DisplayConfig) -> anyhow::Result<Self> {
         let displays = Display::all()
             .map_err(|error| anyhow::anyhow!("Can not get all of displays. {}", error))?;

src-tauri/src/picker/screen.rs

@@ -3,7 +3,6 @@ use std::{io::ErrorKind::WouldBlock, time::Duration, thread};
 
 pub struct Screen {
     capturer: Option<Capturer>,
-    init_error: Option<anyhow::Error>,
     pub width: usize,
     pub height: usize,
 }
@@ -12,16 +11,6 @@ impl Screen {
     pub fn new(capturer: Capturer, width: usize, height: usize) -> Self {
         Self {
             capturer: Some(capturer),
-            init_error: None,
-            width,
-            height,
-        }
-    }
-
-    pub fn new_failed(init_error: anyhow::Error, width: usize, height: usize) -> Self {
-        Self {
-            capturer: None,
-            init_error: Some(init_error),
             width,
             height,
         }

src-tauri/src/rpc/manager.rs

@@ -1,23 +1,42 @@
-use crate::picker::led_color::LedColor;
+use paris::error;
+use tokio::sync::{broadcast, OnceCell};
+use std::fmt::Debug;
 
-use super::mqtt::MqttConnection;
+use crate::{display, models, picker::led_color::LedColor};
-use once_cell::sync::OnceCell;
+
+use super::mqtt::MqttRpc;
 
 pub struct Manager {
-    mqtt: MqttConnection,
+    client: MqttRpc,
+    initialized: bool,
 }
 
 impl Manager {
-    pub fn global() -> &'static Self {
+    pub async fn global() -> &'static Self {
-        static RPC_MANAGER: OnceCell<Manager> = OnceCell::new();
+        static RPC_MANAGER: OnceCell<Manager> = OnceCell::const_new();
 
-        RPC_MANAGER.get_or_init(|| Manager::new())
+        RPC_MANAGER
+            .get_or_init(|| async { Manager::new().await.unwrap() })
+            .await
     }
 
-    pub fn new() -> Self {
+    pub async fn new() -> anyhow::Result<Self> {
-        let mut mqtt = MqttConnection::new();
+        let mqtt = MqttRpc::new().await?;
-        mqtt.initialize();
+        let initialized = match mqtt.initialize().await {
-        Self { mqtt }
+            Ok(_) => true,
+            Err(err) => {
+                error!("initialize for mqtt was failed. {:?}", err);
+                false
+            }
+        };
+        Ok(Self {
+            client: mqtt,
+            initialized,
+        })
+    }
+
+    pub async fn listen(&self) {
+        self.client.listen().await
     }
 
     pub async fn publish_led_colors(&self, colors: &Vec<LedColor>) -> anyhow::Result<()> {
@@ -27,29 +46,18 @@ impl Manager {
             .flatten()
             .collect::<Vec<u8>>();
 
-        self.mqtt
+        self.publish_led_sub_pixels(payload).await
-            .client
-            .publish(
-                "display-ambient-light/desktop/colors",
-                rumqttc::QoS::AtLeastOnce,
-                false,
-                payload,
-            )
-            .await
-            .map_err(|error| anyhow::anyhow!("mqtt publish failed. {}", error))
     }
 
     pub async fn publish_led_sub_pixels(&self, payload: Vec<u8>) -> anyhow::Result<()> {
+        self.client.publish_led_sub_pixels(payload).await
+    }
+    pub async fn publish_desktop_cmd(&self, field: &str, payload: Vec<u8>) -> anyhow::Result<()>
+    {
+        self.client.publish_desktop_cmd(field, payload).await
+    }
 
-        self.mqtt
+    pub fn client(&self) -> &MqttRpc {
-            .client
+        &self.client
-            .publish(
-                "display-ambient-light/desktop/colors",
-                rumqttc::QoS::AtLeastOnce,
-                false,
-                payload,
-            )
-            .await
-            .map_err(|error| anyhow::anyhow!("mqtt publish failed. {}", error))
     }
 }

src-tauri/src/rpc/mod.rs

@@ -1,2 +1,4 @@
-pub mod manager;
+mod manager;
 pub mod mqtt;
+
+pub use manager::*;

src-tauri/src/rpc/mqtt.rs

@@ -1,44 +1,187 @@
-use rumqttc::{AsyncClient, MqttOptions, QoS};
+use crate::{display, models};
-use std::time::Duration;
+use futures::StreamExt;
+use paho_mqtt as mqtt;
+use paris::{error, info, warn};
+use serde_json::json;
+use std::{borrow::Borrow, fmt::Debug, rc::Rc, sync::Arc, time::Duration};
+use tauri::async_runtime::{Mutex, TokioJoinHandle};
 use time::{format_description, OffsetDateTime};
-use tokio::task;
+use tokio::{sync::broadcast, task, time::sleep};
-use tracing::warn;
 
-pub struct MqttConnection {
+const DISPLAY_TOPIC: &'static str = "display-ambient-light/display";
-    pub client: AsyncClient,
+const DESKTOP_TOPIC: &'static str = "display-ambient-light/desktop";
+const DISPLAY_BRIGHTNESS_TOPIC: &'static str = "display-ambient-light/board/brightness";
+const BOARD_SEND_CMD: &'static str = "display-ambient-light/board/cmd";
+
+pub struct MqttRpc {
+    client: mqtt::AsyncClient,
+    change_display_brightness_tx: broadcast::Sender<display::DisplayBrightness>,
+    message_tx: broadcast::Sender<models::CmdMqMessage>,
 }
 
-impl MqttConnection {
+impl MqttRpc {
-    pub fn new() -> Self {
+    pub async fn new() -> anyhow::Result<Self> {
-        let mut options = MqttOptions::new("rumqtt-async", "192.168.31.11", 1883);
+        let client = mqtt::AsyncClient::new("tcp://192.168.31.11:1883")
-        options.set_keep_alive(Duration::from_secs(5));
+            .map_err(|err| anyhow::anyhow!("can not create MQTT client. {:?}", err))?;
 
-        let (client, mut eventloop) = AsyncClient::new(options, 10);
+        client.set_connected_callback(|client| {
-        task::spawn(async move {
+            info!("MQTT server connected.");
-            loop {
+
-                match eventloop.poll().await {
+            client.subscribe("display-ambient-light/board/#", mqtt::QOS_1);
-                    Ok(_) => {}
+
-                    Err(err) => {
+            client.subscribe(format!("{}/#", DISPLAY_TOPIC), mqtt::QOS_1);
-                        println!("MQTT Error Event = {:?}", err);
+        });
+        client.set_connection_lost_callback(|client| {
+            info!("MQTT server connection lost.");
+        });
+        client.set_disconnected_callback(|_, a1, a2| {
+            info!("MQTT server disconnected. {:?} {:?}", a1, a2);
+        });
+
+        let mut last_will_payload = serde_json::Map::new();
+        last_will_payload.insert("message".to_string(), json!("offline"));
+        last_will_payload.insert(
+            "time".to_string(),
+            serde_json::Value::String(
+                OffsetDateTime::now_utc()
+                    .format(
+                        &format_description::parse("[year]-[month]-[day] [hour]:[minute]:[second]")
+                            .unwrap(),
+                    )
+                    .unwrap()
+                    .to_string(),
+            ),
+        );
+
+        let last_will = mqtt::Message::new(
+            format!("{}/status", DESKTOP_TOPIC),
+            serde_json::to_string(&last_will_payload)
+                .unwrap()
+                .as_bytes(),
+            mqtt::QOS_1,
+        );
+
+        let connect_options = mqtt::ConnectOptionsBuilder::new()
+            .keep_alive_interval(Duration::from_secs(5))
+            .will_message(last_will)
+            .automatic_reconnect(Duration::from_secs(1), Duration::from_secs(5))
+            .finalize();
+
+        let token = client.connect(connect_options);
+
+        token.await.map_err(|err| {
+            anyhow::anyhow!(
+                "can not connect MQTT server. wait for connect token failed. {:?}",
+                err
+            )
+        })?;
+
+        let (change_display_brightness_tx, _) =
+            broadcast::channel::<display::DisplayBrightness>(16);
+        let (message_tx, _) = broadcast::channel::<models::CmdMqMessage>(32);
+        Ok(Self {
+            client,
+            change_display_brightness_tx,
+            message_tx,
+        })
+    }
+
+    pub async fn listen(&self) {
+        let change_display_brightness_tx2 = self.change_display_brightness_tx.clone();
+        let message_tx_cloned = self.message_tx.clone();
+
+        let mut stream = self.client.to_owned().get_stream(100);
+
+        while let Some(notification) = stream.next().await {
+            match notification {
+                Some(notification) => match notification.topic() {
+                    DISPLAY_BRIGHTNESS_TOPIC => {
+                        let payload_text = String::from_utf8(notification.payload().to_vec());
+                        match payload_text {
+                            Ok(payload_text) => {
+                                let display_brightness: Result<display::DisplayBrightness, _> =
+                                    serde_json::from_str(payload_text.as_str());
+                                match display_brightness {
+                                    Ok(display_brightness) => {
+                                        match change_display_brightness_tx2.send(display_brightness)
+                                        {
+                                            Ok(_) => {}
+                                            Err(err) => {
+                                                warn!(
+                                                        "can not send display brightness to channel. {:?}",
+                                                        err
+                                                    );
+                                            }
+                                        }
+                                    }
+                                    Err(err) => {
+                                        warn!(
+                                            "can not parse display brightness from payload. {:?}",
+                                            err
+                                        );
+                                    }
+                                }
+                            }
+                            Err(err) => {
+                                warn!("can not parse display brightness from payload. {:?}", err);
+                            }
+                        }
                     }
+                    BOARD_SEND_CMD => {
+                        let payload_text = String::from_utf8(notification.payload().to_vec());
+                        match payload_text {
+                            Ok(payload_text) => {
+                                let message: Result<models::CmdMqMessage, _> =
+                                    serde_json::from_str(payload_text.as_str());
+                                match message {
+                                    Ok(message) => match message_tx_cloned.send(message) {
+                                        Ok(_) => {}
+                                        Err(err) => {
+                                            warn!("can not send message to channel. {:?}", err);
+                                        }
+                                    },
+                                    Err(err) => {
+                                        warn!("can not parse message from payload. {:?}", err);
+                                    }
+                                }
+                            }
+                            Err(err) => {
+                                warn!("can not parse message from payload. {:?}", err);
+                            }
+                        }
+                    }
+                    _ => {}
+                },
+                _ => {
+                    warn!("can not get notification from MQTT server.");
                 }
             }
-        });
+        }
-        Self { client }
     }
 
-    pub fn initialize(&mut self) {
+    pub async fn initialize(&self) -> anyhow::Result<()> {
-        self.subscribe_board();
+        // self.subscribe_board()?;
+        // self.subscribe_display()?;
         self.broadcast_desktop_online();
+        anyhow::Ok(())
     }
 
-    async fn subscribe_board(&self) {
+    fn subscribe_board(&self) -> anyhow::Result<()> {
         self.client
-            .subscribe("display-ambient-light/board/#", QoS::AtMostOnce)
+            .subscribe("display-ambient-light/board/#", mqtt::QOS_1)
-            .await;
+            .wait()
+            .map_err(|err| anyhow::anyhow!("subscribe board failed. {:?}", err))
+            .map(|_| ())
+    }
+    fn subscribe_display(&self) -> anyhow::Result<()> {
+        self.client
+            .subscribe(format!("{}/#", DISPLAY_TOPIC), mqtt::QOS_1)
+            .wait()
+            .map_err(|err| anyhow::anyhow!("subscribe board failed. {:?}", err))
+            .map(|_| ())
     }
 
-    fn broadcast_desktop_online(&mut self) {
+    fn broadcast_desktop_online(&self) {
         let client = self.client.to_owned();
         task::spawn(async move {
             loop {
@@ -46,15 +189,12 @@ impl MqttConnection {
                     .format(&format_description::well_known::Iso8601::DEFAULT)
                 {
                     Ok(now_str) => {
-                        match client
+                        let msg = mqtt::Message::new(
-                            .publish(
+                            "display-ambient-light/desktop/online",
-                                "display-ambient-light/desktop/online",
+                            now_str.as_bytes(),
-                                QoS::AtLeastOnce,
+                            mqtt::QOS_0,
-                                false,
+                        );
-                                now_str.as_bytes(),
+                        match client.publish(msg).await {
-                            )
-                            .await
-                        {
                             Ok(_) => {}
                             Err(error) => {
                                 warn!("can not publish last online time. {}", error)
@@ -69,4 +209,32 @@ impl MqttConnection {
             }
         });
     }
+
+    pub async fn publish_led_sub_pixels(&self, payload: Vec<u8>) -> anyhow::Result<()> {
+        self.client
+            .publish(mqtt::Message::new(
+                "display-ambient-light/desktop/colors",
+                payload,
+                mqtt::QOS_1,
+            ))
+            .await
+            .map_err(|error| anyhow::anyhow!("mqtt publish failed. {}", error))
+    }
+
+    pub fn subscribe_change_display_brightness_rx(
+        &self,
+    ) -> broadcast::Receiver<display::DisplayBrightness> {
+        self.change_display_brightness_tx.subscribe()
+    }
+    pub async fn publish_desktop_cmd(&self, field: &str, payload: Vec<u8>) -> anyhow::Result<()>
+    {
+        self.client
+            .publish(mqtt::Message::new(
+                format!("{}/{}", DESKTOP_TOPIC, field),
+                payload,
+                mqtt::QOS_1,
+            ))
+            .await
+            .map_err(|error| anyhow::anyhow!("mqtt publish failed. {}", error))
+    }
 }