feat: 支持获取和查看板子连接的情况。

2023-04-30 18:44:26 +08:00 · 2023-04-30 18:44:26 +08:00 · 6c90a5e655
commit 6c90a5e655
parent 11045f27d8
6 changed files with 193 additions and 58 deletions
--- a/src-tauri/src/main.rs
+++ b/src-tauri/src/main.rs
@ -419,12 +419,13 @@ async fn main() {
                loop {
                    match UdpRpc::global().await {
                        Ok(udp_rpc) => {
-                            let mut receiver = udp_rpc.clone_boards_change_receiver().await;
+                            let mut receiver = udp_rpc.subscribe_boards_change();
                            loop {
                                if let Err(err) = receiver.changed().await {
                                    error!("boards change receiver changed error: {}", err);
                                    return;
                                }
+                                log::info!("boards changed");

                                let boards = receiver.borrow().clone();

--- a/src-tauri/src/rpc/board_info.rs
+++ b/src-tauri/src/rpc/board_info.rs
@ -1,10 +1,61 @@
-use std::net::{Ipv4Addr};
+use std::{net::Ipv4Addr, time::Duration};

+use paris::warn;
 use serde::{Deserialize, Serialize};
+use tokio::{net::UdpSocket, time::timeout};

 #[derive(Debug, Clone, Eq, PartialEq, Hash, Serialize, Deserialize)]
 pub struct BoardInfo {
-    pub name: String,
+    pub host: String,
    pub address: Ipv4Addr,
    pub port: u16,
-}
+    pub is_online: bool,
+    pub checked_at: Option<std::time::SystemTime>,
+    pub ttl: Option<u128>,
+}
+
+impl BoardInfo {
+    pub fn new(host: String, address: Ipv4Addr, port: u16) -> Self {
+        Self {
+            host,
+            address,
+            port,
+            is_online: false,
+            checked_at: None,
+            ttl: None,
+        }
+    }
+
+    pub async fn check(&mut self) -> anyhow::Result<()> {
+        let socket = UdpSocket::bind("0.0.0.0:0").await?;
+        socket.connect((self.address, self.port)).await?;
+
+        let instant = std::time::Instant::now();
+
+        socket.send(&[1]).await?;
+        let mut buf = [0u8; 1];
+        let recv_future = socket.recv(&mut buf);
+
+        match timeout(Duration::from_secs(5), recv_future).await {
+            Ok(_) => {
+                let ttl = instant.elapsed();
+                log::info!("buf: {:?}", buf);
+                if buf == [1] {
+                    self.is_online = true;
+                } else {
+                    self.is_online = false;
+                }
+                self.ttl = Some(ttl.as_millis());
+            }
+            Err(_) => {
+                warn!("timeout");
+                self.is_online = false;
+                self.ttl = None;
+            }
+        }
+
+        self.checked_at = Some(std::time::SystemTime::now());
+
+        Ok(())
+    }
+}
--- a/src-tauri/src/rpc/udp.rs
+++ b/src-tauri/src/rpc/udp.rs
@ -1,12 +1,11 @@
 use std::{collections::HashSet, sync::Arc, time::Duration};

 use futures::future::join_all;
-use itertools::Itertools;
 use mdns_sd::{ServiceDaemon, ServiceEvent};
 use paris::{error, info, warn};
 use tokio::{
    net::UdpSocket,
-    sync::{watch, Mutex, OnceCell, RwLock},
+    sync::{watch, OnceCell, RwLock},
 };

 use super::BoardInfo;
@ -15,8 +14,7 @@ use super::BoardInfo;
 pub struct UdpRpc {
    socket: Arc<UdpSocket>,
    boards: Arc<RwLock<HashSet<BoardInfo>>>,
-    boards_change_sender: Arc<Mutex<watch::Sender<HashSet<BoardInfo>>>>,
-    boards_change_receiver: Arc<Mutex<watch::Receiver<HashSet<BoardInfo>>>>,
+    boards_change_sender: Arc<watch::Sender<Vec<BoardInfo>>>,
 }

 impl UdpRpc {
@ -36,22 +34,23 @@ impl UdpRpc {
        let socket = UdpSocket::bind("0.0.0.0:0").await?;
        let socket = Arc::new(socket);
        let boards = Arc::new(RwLock::new(HashSet::new()));
-        let (boards_change_sender, boards_change_receiver) = watch::channel(HashSet::new());
-        let boards_change_sender = Arc::new(Mutex::new(boards_change_sender));
-        let boards_change_receiver = Arc::new(Mutex::new(boards_change_receiver));
+        let (boards_change_sender, _) = watch::channel(Vec::new());
+        let boards_change_sender = Arc::new(boards_change_sender);
+
        Ok(Self {
            socket,
            boards,
            boards_change_sender,
-            boards_change_receiver,
        })
    }

    async fn initialize(&self) {
        let shared_self = Arc::new(self.clone());
+
+        let shared_self_for_search = shared_self.clone();
        tokio::spawn(async move {
            loop {
-                match shared_self.search_boards().await {
+                match shared_self_for_search.search_boards().await {
                    Ok(_) => {
                        info!("search_boards finished");
                    }
@ -62,44 +61,63 @@ impl UdpRpc {
                }
            }
        });
+
+        let shared_self_for_check = shared_self.clone();
+        tokio::spawn(async move {
+            shared_self_for_check.check_boards().await;
+        });
+
+        // let shared_self_for_watch = shared_self.clone();
+        // tokio::spawn(async move {
+        //     let mut rx = shared_self_for_watch.clone_boards_change_receiver().await;
+
+        //     // let mut rx  = sub_tx.subscribe();
+        //     // drop(sub_tx);
+        //     while rx.changed().await.is_ok() {
+        //         let boards = rx.borrow().clone();
+        //         info!("boards changed: {:?}", boards);
+        //     }
+        // });
    }

    async fn search_boards(&self) -> anyhow::Result<()> {
        let service_type = "_ambient_light._udp.local.";
        let mdns = ServiceDaemon::new()?;
-        let shared_self = Arc::new(Mutex::new(self.clone()));
        let receiver = mdns.browse(&service_type).map_err(|e| {
            warn!("Failed to browse for {:?}: {:?}", service_type, e);
            e
        })?;
+        let sender = self.boards_change_sender.clone();

        while let Ok(event) = receiver.recv() {
            match event {
                ServiceEvent::ServiceResolved(info) => {
                    info!(
-                    "Resolved a new service: {} host: {} port: {} IP: {:?} TXT properties: {:?}",
-                    info.get_fullname(),
-                    info.get_hostname(),
-                    info.get_port(),
-                    info.get_addresses(),
-                    info.get_properties(),
-                );
+                        "Resolved a new service: {} host: {} port: {} IP: {:?} TXT properties: {:?}",
+                        info.get_fullname(),
+                        info.get_hostname(),
+                        info.get_port(),
+                        info.get_addresses(),
+                        info.get_properties(),
+                    );

-                    let shared_self = shared_self.lock().await;
-                    let mut boards = shared_self.boards.write().await;
+                    let mut boards = self.boards.write().await;

-                    let board = BoardInfo {
-                        name: info.get_fullname().to_string(),
-                        address: info.get_addresses().iter().next().unwrap().clone(),
-                        port: info.get_port(),
-                    };
+                    let board = BoardInfo::new(
+                        info.get_fullname().to_string(),
+                        info.get_addresses().iter().next().unwrap().clone(),
+                        info.get_port(),
+                    );

                    if boards.insert(board.clone()) {
                        info!("added board {:?}", board);
                    }

-                    let sender = self.boards_change_sender.clone().lock_owned().await;
-                    sender.send(boards.clone())?;
+                    let tx_boards = boards.iter().cloned().collect();
+                    drop(boards);
+
+                    sender.send(tx_boards)?;
+                    tokio::task::yield_now().await;
                }
                other_event => {
                    warn!("{:?}", &other_event);
@ -110,32 +128,28 @@ impl UdpRpc {
        Ok(())
    }

-    pub async fn clone_boards_change_receiver(
-        &self,
-    ) -> watch::Receiver<HashSet<BoardInfo>> {
-        let boards_change_receiver = self.boards_change_receiver.clone().lock_owned().await;
-        boards_change_receiver.clone()
+    pub fn subscribe_boards_change(&self) -> watch::Receiver<Vec<BoardInfo>> {
+        self.boards_change_sender.subscribe()
    }

-    pub async fn get_boards(&self) -> HashSet<BoardInfo> {
-        let boards = self.boards.read().await;
-        boards.clone()
+    pub async fn get_boards(&self) -> Vec<BoardInfo> {
+        let boards: tokio::sync::RwLockReadGuard<HashSet<BoardInfo>> = self.boards.read().await;
+        boards.iter().cloned().collect()
    }

    pub async fn send_to_all(&self, buff: &Vec<u8>) -> anyhow::Result<()> {
        let boards = self.get_boards().await;
        let socket = self.socket.clone();
-        
-        let handlers = boards.into_iter()
-        .map(|board| {
+
+        let handlers = boards.into_iter().map(|board| {
            let socket = socket.clone();
            let buff = buff.clone();
            tokio::spawn(async move {
                match socket.send_to(&buff, (board.address, board.port)).await {
-                    Ok(_) => {},
+                    Ok(_) => {}
                    Err(err) => {
-                        error!("failed to send to {}: {:?}", board.name, err);
-                    },
+                        error!("failed to send to {}: {:?}", board.host, err);
+                    }
                }
            })
        });
@ -144,4 +158,31 @@ impl UdpRpc {

        Ok(())
    }
+
+    pub async fn check_boards(&self) {
+        loop {
+            tokio::time::sleep(Duration::from_secs(1)).await;
+
+            let mut boards = self.get_boards().await;
+
+            if boards.is_empty() {
+                info!("no boards found");
+                continue;
+            }
+
+            for board in &mut boards {
+                if let Err(err) = board.check().await {
+                    error!("failed to check board {}: {:?}", board.host, err);
+                }
+            }
+
+            let board_change_sender = self.boards_change_sender.clone();
+            if let Err(err) = board_change_sender.send(boards) {
+                error!("failed to send board change: {:?}", err);
+            } else {
+                info!("send");
+            }
+            drop(board_change_sender);
+        }
+    }
 }
--- a/src/components/info/board-index.tsx
+++ b/src/components/info/board-index.tsx
@ -3,6 +3,7 @@ import { BoardInfo } from '../../models/board-info.model';
 import { listen } from '@tauri-apps/api/event';
 import debug from 'debug';
 import { invoke } from '@tauri-apps/api';
+import { BoardInfoPanel } from './board-info-panel';

 const logger = debug('app:components:info:board-index');

@ -28,19 +29,8 @@ export const BoardIndex: Component = () => {
    <ol class="grid sm:grid-cols-1 md:grid-cols-2 lg:grid-cols-3 p-2 gap-2">
      <For each={boards()}>
        {(board, index) => (
-          <li class="p-2 rounded shadow bg-slate-50 text-gray-800 relative border-2 border-slate-50 hover:border-sky-300 focus:border-sky-300 transition">
-            <dl class="flex">
-              <dt class="w-20">host</dt>
-              <dd class="flex-auto">{board.name}</dd>
-            </dl>
-            <dl class="flex">
-              <dt class="w-20">Ip Addr</dt>
-              <dd class="flex-auto font-mono">{board.address}</dd>
-            </dl>
-            <dl class="flex">
-              <dt class="w-20">Port</dt>
-              <dd class="flex-auto font-mono">{board.port}</dd>
-            </dl>
+          <li class="bg-slate-50 text-gray-800 relative border-2 border-slate-50 hover:border-sky-300 focus:border-sky-300 transition">
+            <BoardInfoPanel board={board} />
            <span class="absolute left-2 -top-3 bg-sky-300 text-white px-1 py-0.5 text-xs rounded-sm font-mono">
              #{index() + 1}
            </span>
--- a/src/components/info/board-info-panel.tsx
+++ b/src/components/info/board-info-panel.tsx
@ -0,0 +1,49 @@
+import { Component, ParentComponent, createMemo } from 'solid-js';
+import { BoardInfo } from '../../models/board-info.model';
+
+type ItemProps = {
+  label: string;
+};
+
+const Item: ParentComponent<ItemProps> = (props) => {
+  return (
+    <dl class="flex">
+      <dt class="w-20">{props.label}</dt>
+      <dd class="flex-auto">{props.children}</dd>
+    </dl>
+  );
+};
+
+export const BoardInfoPanel: Component<{ board: BoardInfo }> = (props) => {
+  const ttl = createMemo(() => {
+    if (!props.board.is_online) {
+      return '--';
+    }
+
+    if (props.board.ttl == null) {
+      return 'timeout';
+    }
+
+    return (
+      <>
+        <span class="font-mono">{props.board.ttl.toFixed(0)}</span> ms
+      </>
+    );
+  });
+
+  return (
+    <section class="p-2 rounded shadow">
+      <Item label="Host">{props.board.host}</Item>
+      <Item label="Ip Addr">
+        <span class="font-mono">{props.board.address}</span>
+      </Item>
+      <Item label="Port">
+        <span class="font-mono">{props.board.port}</span>
+      </Item>
+      <Item label="Status">
+        <span class="font-mono">{props.board.is_online ? 'Online' : 'Offline'}</span>
+      </Item>
+      <Item label="TTL">{ttl()}</Item>
+    </section>
+  );
+};
--- a/src/models/board-info.model.ts
+++ b/src/models/board-info.model.ts
@ -1,5 +1,8 @@
 export type BoardInfo = {
-  name: string;
+  host: string;
  address: string;
  port: number;
+  ttl: number;
+  is_online: boolean;
+  checked_at: Date;
 };