Websocket now reconnects after disconnect

Cargo.lock

@@ -628,6 +628,7 @@ dependencies = [
  "futures-util",
  "serde",
  "serde_json",
+ "tokio",
  "uuid",
 ]
 
@@ -1357,9 +1358,21 @@ dependencies = [
  "pin-project-lite",
  "signal-hook-registry",
  "socket2 0.6.5",
+ "tokio-macros",
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "tokio-macros"
+version = "2.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.7.19"

Cargo.toml

@@ -11,6 +11,7 @@ dotenvy = "0.15.7"
 futures-util = "0.3.33"
 serde = { version = "1.0.229", features = ["derive"]}
 serde_json = "1.0.151"
+tokio = { version="1.53.1", features = ["macros", "rt-multi-thread", "sync", "time"]}
 uuid = { version = "1.24.0", features = ["v4", "serde"]}
 
 [profile.release]

src/controllers/create_first_admin.rs

@@ -0,0 +1,6 @@
+use actix_web::{HttpResponse, web, post};
+
+#[post("/admin/first")]
+pub async fn route(
+    
+)

src/main.rs

@@ -1,4 +1,6 @@
 use actix_web::{HttpServer, App, middleware, web, rt};
+use awc::ws::Message;
+use tokio::sync::mpsc;
 use std::sync::Mutex;
 
 mod routes;
@@ -16,18 +18,18 @@ async fn main() -> Result<(), std::io::Error> {
 
     //Read User Data
     let users = Mutex::new(users::User::read());
-    let data = web::Data::new(users);
+    let users_data = web::Data::new(users);
 
-    //Connect to HA
-    let ha_connection = websocket::connect_to_ha(ha_ip, ha_token)
-        .await
-        .expect("Failed to start websocket");
-    rt::spawn(websocket::listen(ha_connection));
+    //Connect to HA via websocket
+    let (outbound_tx, outbound_rx) = mpsc::unbounded_channel::<Message>();
+    let ha_sender = web::Data::new(outbound_tx);
+    rt::spawn(websocket::listen(ha_ip, ha_token, outbound_rx));
 
     //Create HTTP server
     HttpServer::new(move || {
         App::new()
-            .app_data(data.clone())
+            .app_data(users_data.clone())
+            .app_data(ha_sender.clone())
             .wrap(middleware::Compress::default())
             .configure(routes::config)
     })

src/websocket.rs

@@ -2,6 +2,7 @@ use awc::{Client, BoxedSocket, ws::{Frame, Message, Codec}};
 use actix_codec::Framed;
 use futures_util::{SinkExt, StreamExt};
 use serde_json::{Value, from_str, json};
+use tokio::sync::mpsc;
 
 pub async fn connect_to_ha(
     ip: String,
@@ -51,37 +52,64 @@ pub async fn connect_to_ha(
     Err("Websocket closed before authentication completed".into())
 }
 
-pub async fn listen(mut connection: Framed<BoxedSocket, Codec>) {
-    let mut message_id = 1;
+pub async fn listen(
+    ip: String,
+    token: String,
+    mut outbound_rx: mpsc::UnboundedReceiver<Message>
+) {
+    loop {
+        let mut connection = match connect_to_ha(ip.clone(), token.clone()).await {
+            Ok(c) => c,
+            Err(e) => {
+                eprintln!("Failed to connect to HA: {e}. Retrying in 10s...");
+                tokio::time::sleep(std::time::Duration::from_secs(10)).await;
+                continue;
+            }
+        };
 
-    while let Some(Ok(msg)) = connection.next().await {
-        match msg {
-            Frame::Text(bytes) => {
-                let text = match String::from_utf8(bytes.to_vec()) {
-                    Ok(t) => t,
-                    Err(_) => continue
-                };
-                let parsed: Value = match from_str(&text) {
-                    Ok(v) => v,
-                    Err(_) => continue
-                };
+        loop {
+            tokio::select! {
+                msg = connection.next() => {
+                    match msg {
+                        Some(Ok(Frame::Text(bytes))) => {
+                            let text = match String::from_utf8(bytes.to_vec()) {
+                                Ok(t) => t,
+                                Err(_) => continue
+                            };
+                            let parsed: Value = match from_str(&text) {
+                                Ok(v) => v,
+                                Err(_) => continue
+                            };
 
-                if parsed["type"] == "result" {
-                    println!("result");
-                    println!("{:#?}", parsed);
+                            if parsed["type"] == "result" {
+                                println!("result");
+                                println!("{:#?}", parsed);
+                            }
+                        }
+                        Some(Ok(Frame::Ping(bytes))) => {
+                            let _ = connection.send(Message::Pong(bytes)).await;
+                        }
+                        Some(Ok(Frame::Close(reason))) => {
+                            println!("HA closed connection: {:?}", reason);
+                            break;
+                        }
+                        Some(Err(e)) => {
+                            eprintln!("Websocket error: {e}");
+                            break;
+                        }
+                        None => break,
+                        _ => {}
+                    }
+                }
+                Some(out_msg) = outbound_rx.recv() => {
+                    if let Err(e) = connection.send(out_msg).await {
+                        eprintln!("Failed to send outbound message: {e}");
+                    }
                 }
-            },
-            Frame::Ping(bytes) => {
-                let _ = connection.send(Message::Pong(bytes)).await;
-            },
-            Frame::Close(reason) => {
-                println!("HA closed connection {:?}", reason);
-                
-                
-            },
-            _ => {}
+            }
         }
-    }
 
-    println!("HA websocket connection closed");
+        println!("HA connection closed, reconnecting in 10s...");
+        tokio::time::sleep(std::time::Duration::from_secs(10)).await;
+    }
 }

ui/adminCreate/index.js

@@ -1,5 +1,26 @@
 export default {
+    form: null,
+
     display(){
-        console.log("displaying admin creation");
+        if(this.form === null){
+            this.form = document.querySelector("#adminCreatePage form");
+            this.form.addEventListener("submit", (event)=>{
+                this.submit(event);
+            });
+        }
+
+        document.getElementById("adminCreateName").focus();
+    },
+
+    submit(event){
+        event.preventDefault();
+
+        const data = {
+            name: document.getElementById("adminCreateName").value,
+            email: document.getElementById("adminCreateEmail").value,
+            password: document.getElementById("adminCreatePass").value,
+            confirm_password: document.getElementById("adminCreatePassConf").value
+        };
+        console.log(data);
     }
 }