feature: ping from server to clients added

Signed-off-by: Leonid Kaganov <[email protected]>

Author: lleokaganov

000rename.sh

@@ -0,0 +1,18 @@
+mv "Cargo.lock"  "Cargo.lock"
+mv "Cargo.toml"  "Cargo.toml"
+mv "DOCKER.sh"   "DOCKER.sh"
+mv "Dockerfile"  "Dockerfile"
+mv "GO.sh"       "GO.sh"
+mv "LICENSE"     "LICENSE"
+mv "README.md"   "README.md"
+mv "TEST.sh"     "TEST.sh"
+mv "TEST_WS.sh"  "TEST_WS.sh"
+mv "TODO.txt"    "TODO.txt"
+mv "client"      "client"
+mv "commit.sh"   "commit.sh"
+mv "lleo"        "lleo"
+mv "policy.repo" "policy.repo"
+mv "scripts"     "scripts"
+mv "src"         "src"
+mv "target"      "target"
+mv "tests"       "tests"

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "hulypulse"
-version = "0.2.0"
+version = "0.2.1"
 edition = "2024"
 
 [dependencies]

Cargo.toml

@@ -110,7 +110,11 @@ <h1>WebSocket JSON Tester</h1>
 
     ws.onopen = () => {
 	pr("clear");
-	pr("✅ WebSocket connected.");
+	pr("✅ WebSocket connected");
+    };
+
+    ws.onclose = () => {
+	pr("❌ WebSocket closed");
     };
 
     ws.onmessage = (e) => {

scripts/TEST.html

@@ -58,6 +58,7 @@ pub struct Config {
     pub no_authorization: bool,
 
     pub heartbeat_timeout: u64,
+    pub ping_timeout: u64,
 
     pub policy_file: Option<String>,
 }

src/config.rs

@@ -13,6 +13,7 @@ backend = "redis"
 no_authorization = false
 
 heartbeat_timeout = 90
+ping_timeout = 30
 
 # optional settings
 # max_size = 100

src/config/default.toml

@@ -317,14 +317,15 @@ pub async fn handler(
             hub_state.write().await.renew_heartbeat(session_id);
 
             match msg {
-                // actix_ws::Message::Text(text) => {
-                //     session.text(format!("echo: {}", text)).await.ok();
-                // }
                 actix_ws::Message::Ping(bytes) => {
                     session.pong(&bytes).await.ok();
                     continue;
                 }
 
+                actix_ws::Message::Pong(_) => {
+                    continue;
+                }
+
                 actix_ws::Message::Text(text) if text == "ping" => {
                     let _ = session.text("pong").await;
                     continue;

src/handlers_ws.rs

@@ -67,25 +67,30 @@ pub struct HubState {
     sessions: HashMap<SessionId, actix_ws::Session>,
     subs: HashMap<String, HashSet<SessionId>>,
     heartbeats: HashMap<SessionId, std::time::Instant>,
+    serverping: HashMap<SessionId, std::time::Instant>,
 }
 
 impl HubState {
     pub fn renew_heartbeat(&mut self, session_id: SessionId) {
         if self.sessions.contains_key(&session_id) {
             let now = std::time::Instant::now();
             self.heartbeats.insert(session_id, now);
+            self.serverping.insert(session_id, now);
         }
     }
 
     pub fn connect(&mut self, session_id: SessionId, session: actix_ws::Session) {
         self.sessions.insert(session_id, session);
         self.heartbeats
             .insert(session_id, std::time::Instant::now());
+        self.serverping
+            .insert(session_id, std::time::Instant::now());
     }
 
     pub fn disconnect(&mut self, session_id: SessionId) {
         self.sessions.remove(&session_id);
         self.heartbeats.remove(&session_id);
+        self.serverping.remove(&session_id);
         self.subs.retain(|_, ids| {
             ids.remove(&session_id);
             !ids.is_empty()
@@ -192,10 +197,11 @@ pub fn check_heartbeat(hub_state: Arc<RwLock<HubState>>) {
             ticker.tick().await;
 
             let now = std::time::Instant::now();
-            let timeout = std::time::Duration::from_secs(CONFIG.heartbeat_timeout);
-            let timelimit = now - timeout;
+            let timelimit = now - std::time::Duration::from_secs(CONFIG.heartbeat_timeout);
+            let pinglimit = now - std::time::Duration::from_secs(CONFIG.ping_timeout);
 
             let hub = hub_state.read().await;
+
             let expired: Vec<actix_ws::Session> = hub
                 .heartbeats
                 .iter()
@@ -208,6 +214,18 @@ pub fn check_heartbeat(hub_state: Arc<RwLock<HubState>>) {
                 })
                 .collect();
 
+            let to_ping: Vec<SessionId> = hub
+                .serverping
+                .iter()
+                .filter_map(|(&sid, &last_ping)| {
+                    if last_ping < pinglimit {
+                        Some(sid)
+                    } else {
+                        None
+                    }
+                })
+                .collect();
+
             drop(hub);
 
             if !expired.is_empty() {
@@ -216,6 +234,17 @@ pub fn check_heartbeat(hub_state: Arc<RwLock<HubState>>) {
                     let _ = addr.clone().close(None).await;
                 }
             }
+
+            if !to_ping.is_empty() {
+                let mut hub = hub_state.write().await;
+                for sid in &to_ping {
+                    if let Some(session) = hub.sessions.get_mut(sid) {
+                        let _ = session.ping(&[]).await;
+                    }
+                    hub.serverping.insert(*sid, now);
+                }
+                drop(hub);
+            }
         }
     });
 }