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,10 +8,10 @@ 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;
+use tasks::can_data::can_data_handler::run_can_data_handler;
 
 use crate::tasks::api_handler::run_api_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,15 +1,17 @@
+use std::sync::Arc;
 
-use tokio::{select};
+use tokio::{select, spawn};
 use tokio::sync::{RwLock, broadcast};
 
-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 super::influx_handler::run_influx_handler;
+use super::live_data_handler::run_live_data_handler;
+use crate::tasks::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(), clients));
 
     loop {
         select! {
@@ -44,13 +46,6 @@ pub async fn run_can_data_handler(
                 break;
             }
             Ok(can_payload) = can_queue_rx.recv() => {
-                // let can_message = match can_db.get_message_by_id(can_payload.can_id) {
-                //     Ok(c) => c,
-                //     Err(_) => {
-                //         eprintln!("Unknown CAN ID: {:X}", can_payload.can_id);
-                //         continue;
-                //     },
-                // };
 
                 let decoded_signals = can_db.unpack(
                     can_payload.can_id, 
@@ -59,16 +54,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
+                    if !decoded_signal_tx.send(signal).is_ok() {
+                        eprintln!("Parsed can data signal consumers are all closed");
+                        break;
                     }
                 }
             }
         }
     }
+    
+    let _ = influx_handler_task.await;
+    let _ = live_data_handler_task.await;
     println!("CAN data handler task ended.");
-
-    // influx_handler_task.await;
-    // live_data_handler_task.await;
 }
 

…st_backend/src/tasks/can_data_handler.rs …d/src/tasks/can_data/can_data_handler.rssoftware/tracksight/rust_backend/src/tasks/can_data_handler.rs renamed to software/tracksight/rust_backend/src/tasks/can_data/can_data_handler.rs software/tracksight/rust_backend/src/tasks/can_data_handler.rs renamed to software/tracksight/rust_backend/src/tasks/can_data/can_data_handler.rs

@@ -6,13 +6,13 @@ use tokio::select;
 use crate::config::CONFIG;
 use crate::tasks::telem_message::CanPayload;
 
-use jsoncan_rust::can_database::CanMessage;
+use jsoncan_rust::can_database::{CanMessage, DecodedSignal};
 
 /**
  * After serial_handler parses the can messages,
  * this task consumes the messages and writes them to influxdb
  */
-pub async fn run_influx_handler(mut shutdown_signal: Receiver<()>, mut can_queue_receiver: Receiver<()>) {
+pub async fn run_influx_handler(mut shutdown_signal: Receiver<()>, mut can_queue_receiver: Receiver<DecodedSignal>) {
     println!("Influx task started.");
 
     let influx_client = Client::new(

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

@@ -1,17 +1,21 @@
-use tokio::sync::broadcast::Receiver;
-use influxdb2::Client;
+use std::sync::Arc;
+
+use tokio::sync::{RwLock, broadcast::Receiver};
 use tokio::select;
 
-use crate::config::CONFIG;
-use crate::tasks::telem_message::CanPayload;
+use crate::tasks::client_api::clients::Clients;
 
-use jsoncan_rust::can_database::CanMessage;
+use jsoncan_rust::can_database::DecodedSignal;
 
 /**
  * After serial_handler parses the can messages,
  * this task will handle "forwarding" live data to clients via sockets
  */
-pub async fn run_live_data_handler(mut shutdown_signal: Receiver<()>, mut can_signals_rx: Receiver<()>) {
+pub async fn run_live_data_handler(
+    mut shutdown_signal: Receiver<()>, 
+    mut can_signals_rx: Receiver<DecodedSignal>,
+    clients: Arc<RwLock<Clients>>
+) {
     println!("Live data task started.");
 
     loop {
@@ -20,9 +24,21 @@ pub async fn run_live_data_handler(mut shutdown_signal: Receiver<()>, mut can_si
                 println!("Shutting down live data handler task.");
                 break;
             }
-            _ = can_signals_rx.recv() => {
+            Ok(signal) = can_signals_rx.recv() => {
                 // todo should also probably check for closed channels and close thread
                 // todo handle
+                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
+                    let _ = c.get_client_socket(client_id).clone().unwrap().emit(
+                        "data", 
+                        &serde_json::json!({
+                            "name": signal.name,
+                            "value": signal.value,
+                            "timestamp": signal.timestamp,
+                        })
+                    );
+                }
             }
         }
     }

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

@@ -1,2 +1,3 @@
 pub mod live_data_handler;
-pub mod influx_handler;
+pub mod influx_handler;
+pub mod can_data_handler;