use std::str::FromStr; use std::net::IpAddr; use std::sync::Arc; use axum::extract::{ws::WebSocket}; use axum::extract::ws::Message; use dashmap::DashMap; use time::{Duration, Instant}; use tokio::sync::broadcast::{Sender}; use tracing::{debug, error, trace}; use crate::AppState; use crate::config::SETTINGS; pub type PingMap = DashMap; #[derive(Debug, Clone)] pub struct PingValue { pub ip: String, pub online: bool } pub async fn spawn(tx: Sender, ip: String, uuid: String, ping_map: &PingMap) { let timer = Instant::now(); let payload = [0; 8]; let ping_ip = IpAddr::from_str(&ip).expect("bad ip"); let mut msg: Option = None; while msg.is_none() { let ping = surge_ping::ping( ping_ip, &payload ).await; if let Err(ping) = ping { let ping_timeout = matches!(ping, surge_ping::SurgeError::Timeout { .. }); if !ping_timeout { error!("{}", ping.to_string()); msg = Some(BroadcastCommands::Error(uuid.clone())); } if timer.elapsed() >= Duration::minutes(SETTINGS.get_int("pingtimeout").unwrap_or(10)) { msg = Some(BroadcastCommands::Timeout(uuid.clone())); } } else { let (_, duration) = ping.map_err(|err| error!("{}", err.to_string())).expect("fatal error"); debug!("ping took {:?}", duration); msg = Some(BroadcastCommands::Success(uuid.clone())); }; } let msg = msg.expect("fatal error"); let _ = tx.send(msg.clone()); if let BroadcastCommands::Success(..) = msg { ping_map.insert(uuid.clone(), PingValue { ip: ip.clone(), online: true }); tokio::time::sleep(tokio::time::Duration::from_secs(60)).await; } trace!("remove {} from ping_map", uuid); ping_map.remove(&uuid); } #[derive(Clone, Debug, PartialEq)] pub enum BroadcastCommands { Success(String), Timeout(String), Error(String), } pub async fn status_websocket(mut socket: WebSocket, state: Arc) { trace!("wait for ws message (uuid)"); let msg = socket.recv().await; let uuid = msg.unwrap().unwrap().into_text().unwrap(); trace!("Search for uuid: {:?}", uuid); let device_exists = state.ping_map.contains_key(&uuid); match device_exists { true => { let _ = socket.send(process_device(state.clone(), uuid).await).await; }, false => { debug!("didn't find any device"); let _ = socket.send(Message::Text(format!("notfound_{}", uuid))).await; }, }; let _ = socket.close().await; } async fn process_device(state: Arc, uuid: String) -> Message { let pm = state.ping_map.clone().into_read_only(); let device = pm.get(&uuid).expect("fatal error"); debug!("got device: {} (online: {})", device.ip, device.online); match device.online { true => { debug!("already started"); Message::Text(format!("start_{}", uuid)) }, false => { loop{ trace!("wait for tx message"); let message = state.ping_send.subscribe().recv().await.expect("fatal error"); trace!("got message {:?}", message); return match message { BroadcastCommands::Success(msg_uuid) => { if msg_uuid != uuid { continue; } trace!("message == uuid success"); Message::Text(format!("start_{}", uuid)) }, BroadcastCommands::Timeout(msg_uuid) => { if msg_uuid != uuid { continue; } trace!("message == uuid timeout"); Message::Text(format!("timeout_{}", uuid)) }, BroadcastCommands::Error(msg_uuid) => { if msg_uuid != uuid { continue; } trace!("message == uuid error"); Message::Text(format!("error_{}", uuid)) } } } } } }