RAHHHHHH frontend receives from socket

Author: j-ackyao

software/tracksight/rust_backend/Cargo.lock

@@ -14,5 +14,6 @@ socketioxide = "0.18.1"
 influxdb2 = "0.5"
 dotenv = "0.15"
 crc = "3.0.0"
+serde_json = "1.0.145"
 
 jsoncan_rust = { path = "../../../scripts/code_generation/jsoncan-rust" }

software/tracksight/rust_backend/Cargo.toml

@@ -1,4 +1,5 @@
 use ctrlc;
+use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use tokio::sync::RwLock;
 use tokio::sync::broadcast;
@@ -7,7 +8,7 @@ use tokio::task::{JoinSet};
 mod config;
 mod tasks;
 use tasks::telem_message::CanPayload;
-use tasks::client_api::subscriptions::Subscriptions;
+use tasks::client_api::clients::Clients;
 
 use tasks::serial_handler::run_serial_task;
 use tasks::can_data_handler::run_can_data_handler;
@@ -41,11 +42,11 @@ async fn main() {
 
     // track which clients subscribe to which signals
     // maps signal name to client ids
-    let subscriptions = RwLock::new(Subscriptions::new());
+    let clients = Arc::new(RwLock::new(Clients::new()));
 
     // start tasks
-    tasks.spawn(run_api_handler(shutdown_rx.resubscribe()));
-    tasks.spawn(run_can_data_handler(shutdown_rx.resubscribe(), can_queue_rx, subscriptions));
+    tasks.spawn(run_api_handler(shutdown_rx.resubscribe(), clients.clone()));
+    tasks.spawn(run_can_data_handler(shutdown_rx.resubscribe(), can_queue_rx, clients.clone()));
     tasks.spawn(run_serial_task(shutdown_rx.resubscribe(), can_queue_tx));
 
     // wait for tasks to clean up

software/tracksight/rust_backend/src/main.rs

@@ -1,32 +1,40 @@
+
+use std::sync::Arc;
 use axum::response::IntoResponse;
 use axum::serve::Listener;
 use tokio::select;
-use tokio::sync::broadcast;
+use tokio::sync::{RwLock, broadcast};
 use tokio::net::TcpListener;
 use axum::{Router, routing::get};
 use axum::extract::ws::{WebSocket, WebSocketUpgrade};
 use socketioxide::{SocketIo, extract::SocketRef};
 use crate::config::CONFIG;
+use crate::tasks::client_api::clients::Clients;
 
 
 
-pub async fn run_api_handler(mut shutdown_rx: broadcast::Receiver<()>) {
+pub async fn run_api_handler(mut shutdown_rx: broadcast::Receiver<()>, clients: Arc<RwLock<Clients>>) {
     let addr = format!("0.0.0.0:{}", CONFIG.backend_port);
     let listener = TcpListener::bind(addr).await.unwrap();
 
     let (socket_layer, io) = SocketIo::new_layer();
 
+    // default socketio endpoint
     io.ns("/", |socket: SocketRef| async move {
-        println!("connected sid={}", socket.id);
+        let client_id = socket.id.to_string();
+        println!("{} connected", client_id);
+
+        clients.write().await.add_client(&client_id, &socket);
 
-        socket.on_disconnect(|| async {
-            println!("disconnected");
+        socket.on_disconnect(async move || {
+            clients.write().await.remove_client(&client_id);
+            println!("{} disconnected", client_id);
         });
+
     });
 
     let app: Router<> = Router::new()
         .layer(socket_layer);
-        // .route("/socket.io", get(handle_connection));
 
     select! {
         _ = shutdown_rx.recv() => {
@@ -37,15 +45,3 @@ pub async fn run_api_handler(mut shutdown_rx: broadcast::Receiver<()>) {
         }
     }
 }
-
-// async fn handle_connection(ws: WebSocketUpgrade) -> impl IntoResponse {
-//     let _ = ws.on_upgrade(async |mut socket: WebSocket| {
-//         println!("client connected");
-
-//         while let Some(Ok(_)) = socket.recv().await {
-//             // ignore
-//         }
-
-//         println!("client disconnected")
-//     });
-// }

software/tracksight/rust_backend/src/tasks/api_handler.rs

@@ -1,5 +1,4 @@
 use tokio::sync::broadcast::Receiver;
-use influxdb2::Client;
 use tokio::select;
 
 use crate::config::CONFIG;

software/tracksight/rust_backend/src/tasks/can_data/live_data_handler.rs

@@ -1,3 +1,4 @@
+use std::sync::Arc;
 
 use tokio::{select};
 use tokio::sync::{RwLock, broadcast};
@@ -6,10 +7,11 @@ use super::can_data::influx_handler::run_influx_handler;
 use super::can_data::live_data_handler::run_live_data_handler;
 use super::telem_message::CanPayload;
 use crate::config::CONFIG;
-use crate::tasks::client_api::subscriptions::Subscriptions;
+use crate::tasks::client_api::clients::Clients;
 
 use jsoncan_rust::parsing::JsonCanParser;
 use jsoncan_rust::can_database::CanDatabase;
+use jsoncan_rust::can_database::DecodedSignal;
 
 /**
  * Consumes from serial handler
@@ -18,7 +20,7 @@ use jsoncan_rust::can_database::CanDatabase;
 pub async fn run_can_data_handler(
     mut shutdown_rx: broadcast::Receiver<()>, 
     mut can_queue_rx: broadcast::Receiver<CanPayload>,
-    subscriptions: RwLock<Subscriptions>
+    clients: Arc<RwLock<Clients>>
 ) {
     println!("CAN data handler task started.");
 
@@ -31,11 +33,11 @@ pub async fn run_can_data_handler(
         }
     };
 
-    // let (parsed_can_tx, _) = broadcast::channel::<can_database::CanMessage>(32);
+    let (decoded_signal_tx, _) = broadcast::channel::<DecodedSignal>(32);
 
     // parsed can signal consumers
-    // let influx_handler_task = spawn(run_influx_handler(shutdown_rx.resubscribe(), parsed_can_tx.subscribe()));
-    // let live_data_handler_task = spawn(run_live_data_handler(shutdown_rx.resubscribe(), parsed_can_tx.subscribe()));
+    // let influx_handler_task = spawn(run_influx_handler(shutdown_rx.resubscribe(), decoded_signal_tx.subscribe()));
+    // let live_data_handler_task = spawn(run_live_data_handler(shutdown_rx.resubscribe(), decoded_signal_tx.subscribe()));
 
     loop {
         select! {
@@ -59,8 +61,17 @@ pub async fn run_can_data_handler(
                 );
 
                 for signal in decoded_signals {
-                    for client_id in subscriptions.read().await.get_clients_of_signal(signal.name) {
-                        // todo emit to client
+                    let c = clients.read().await;
+                    for client_id in c.get_clients_of_signal(&signal.name) {
+                        // todo unwrap or handle none case, which would be really weird
+                        c.get_client_socket(client_id).clone().unwrap().emit(
+                            "data", 
+                            &serde_json::json!({
+                                "name": signal.name,
+                                "value": signal.value,
+                                "timestamp": can_payload.can_timestamp,
+                            })
+                        );
                     }
                 }
             }