Complete the front/back end connection

src/controllers/ws_connect.rs

@@ -6,7 +6,13 @@ use crate::{
     users::User,
     ws_registry::ClientRegistry
 };
-use std::sync::Mutex;
+use std::{
+    sync::{Arc, Mutex},
+    time::{Duration, Instant},
+};
+
+const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30);
+const CLIENT_TIMEOUT: Duration = Duration::from_secs(60);
 
 #[get("/ws")]
 pub async fn route(
@@ -26,7 +32,9 @@ pub async fn route(
     };
 
     let (tx, mut rx) = mpsc::unbounded_channel::<String>();
-    let client_id = registry.register(tx, user.id);
+    let conn_id = Uuid::new_v4();
+    let client_id = registry.register(tx, conn_id);
+    let last_pong = Arc::new(Mutex::new(Instant::now()));
 
     let mut push_session = session.clone();
     actix_web::rt::spawn(async move {
@@ -37,16 +45,56 @@ pub async fn route(
         }
     });
 
+    let mut heartbeat_session = session.clone();
+    let last_pong_hb = Arc::clone(&last_pong);
+    actix_web::rt::spawn(async move {
+        let mut interval = actix_web::rt::time::interval(HEARTBEAT_INTERVAL);
+
+        loop {
+            interval.tick().await;
+
+            if last_pong_hb.lock().unwrap().elapsed() > CLIENT_TIMEOUT {
+                let _ = heartbeat_session.close(None).await;
+                break;
+            }
+
+            if heartbeat_session.ping(b"").await.is_err() {
+                break;
+            }
+        }
+    });
+
     let registry = registry.clone();
     actix_web::rt::spawn(async move {
         use futures_util::StreamExt;
 
         while let Some(Ok(msg)) = msg_stream.next().await {
-            if let actix_ws::Message::Text(text) = msg {
-                let _ = session.text(text).await;
+            match msg {
+                actix_ws::Message::Ping(bytes) => {
+                    *last_pong.lock().unwrap() = Instant::now();
+                    if session.pong(&bytes).await.is_err() {
+                        break;
+                    }
+                }
+                actix_ws::Message::Pong(_) => {
+                    *last_pong.lock().unwrap() = Instant::now();
+                }
+                actix_ws::Message::Text(text) => {
+                    *last_pong.lock().unwrap() = Instant::now();
+                    if session.text(text).await.is_err() {
+                        break;
+                    }
+                }
+                actix_ws::Message::Close(reason) => {
+                    let _ = session.close(reason).await;
+                    registry.unregister(&client_id);
+                    return;
+                }
+                _ => break,
             }
         }
 
+        let _ = session.close(None).await;
         registry.unregister(&client_id);
     });
 

ui/dashboard/index.js

@@ -11,6 +11,12 @@ export default {
                     headers: {"Content-Type": "application/json"}
                 });
             });
+
+            console.log(location.protocol);
+            window.ws = new WebSocket(`${location.protocol === "https:" ? "wss" : "ws"}://${location.host}/ws`);
+            ws.addEventListener("open", ()=>{
+                ws.send("hello");
+            });
         }
     }
 }