subscriptions, todo app

Author: j-ackyao

software/tracksight/rust_backend/Cargo.lock

@@ -11,5 +11,6 @@ tokio-serial = "5"
 influxdb2 = "0.5"
 dotenv = "0.15"
 crc = "3.0.0"
+rust_socketio = "0.6.0"
 
 jsoncan_rust = { path = "../../../scripts/code_generation/jsoncan-rust" }

software/tracksight/rust_backend/Cargo.toml

@@ -14,6 +14,7 @@ todo for backend
 [ ] socket clients to publish data
 [ ] proper error handling in backend and using optimal threading or whatnot
 [ ] implementing api endpoints
+[ ] figure out the time offset basetime thing
 [ ] implementing ntp and base time syncing
 ...
 

software/tracksight/rust_backend/README.md

@@ -1,14 +1,14 @@
 use ctrlc;
 use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::Arc;
-use dashmap::DashMap;
+use tokio::sync::RwLock;
 use tokio::sync::broadcast::channel;
 use tokio::task::{JoinSet};
 
 
 mod config;
 mod tasks;
 use tasks::telem_message::CanPayload;
+use tasks::api::subscriptions::Subscriptions;
 
 use tasks::serial_handler::run_serial_task;
 use tasks::can_data_handler::run_can_data_handler;
@@ -40,10 +40,10 @@ async fn main() {
 
     // track which clients subscribe to which signals
     // maps signal name to client ids
-    let subscribers = Arc::new(DashMap<String, Vec<String>>);
+    let subscriptions = RwLock::new(Subscriptions::new());
 
     // start tasks
-    tasks.spawn(run_can_data_handler(shutdown_rx.resubscribe(), can_queue_rx));
+    tasks.spawn(run_can_data_handler(shutdown_rx.resubscribe(), can_queue_rx, subscriptions));
     tasks.spawn(run_serial_task(shutdown_rx.resubscribe(), can_queue_tx));
 
     // wait for tasks to clean up

software/tracksight/rust_backend/src/main.rs

@@ -0,0 +1 @@
+pub mod subscriptions;

software/tracksight/rust_backend/src/tasks/api/mod.rs

@@ -0,0 +1,176 @@
+use std::collections::{HashMap, HashSet};
+
+pub struct Subscriptions {
+    // Clients and their set of signals subscribed
+    // client -> signal
+    client_to_signal: HashMap<String, HashSet<String>>,
+    // Signals and its subscribed clients
+    // signal -> client
+    signal_to_client: HashMap<String, HashSet<String>>
+}
+
+impl Subscriptions {
+    pub fn new() -> Self {
+        return Self { 
+            client_to_signal: HashMap::new(), 
+            signal_to_client: HashMap::new() 
+        };
+    }
+
+    /**
+     * Subscribe client to signal
+     */
+    pub fn subscribe_client_to_signal(&mut self, client: String, signal: String) {
+        // check if client exists, and add
+        if let Some(client_signals) = self.client_to_signal.get_mut(&client) {
+            client_signals.insert(signal.clone());
+        } else {
+            let mut new_client_signals = HashSet::new();
+            new_client_signals.insert(signal.clone());
+            self.client_to_signal.insert(client.clone(), new_client_signals);
+        }
+        
+        // check if signal exists and add signal
+        if let Some(signal_clients) = self.signal_to_client.get_mut(&signal) {
+            signal_clients.insert(client.clone());
+        } else {
+            let mut new_signal_clients = HashSet::new();
+            new_signal_clients.insert(client.clone());
+            self.signal_to_client.insert(signal.clone(), new_signal_clients);
+        }
+    }
+
+    /**
+     * Remove client from signal
+     */
+    pub fn unsubscribe_client_from_signal(&mut self, client: String, signal: String) {
+        // check if client exists
+        if let Some(client_signals) = self.client_to_signal.get_mut(&client) {
+            client_signals.remove(&signal);
+
+            if client_signals.len() <= 0 {
+                self.client_to_signal.remove(&client);
+            }
+        }
+        
+        // check if signal exists and add signal
+        if let Some(signal_clients) = self.signal_to_client.get_mut(&signal) {
+            signal_clients.remove(&client);
+
+            if signal_clients.len() <= 0 {
+                self.signal_to_client.remove(&signal);
+            }
+        }
+    }
+
+    pub fn get_clients_of_signal(&self, signal: String) -> Vec<&String> {
+        return self.signal_to_client
+            .get(&signal)
+            .map(|clients| clients.iter().collect())
+            .unwrap_or_default();
+    }
+
+    /**
+     * Check if client is subscribed to signal
+     */
+    pub fn is_client_subscribed_to(&self, client: String, signal: String) -> bool {
+        self.client_to_signal
+            .get(&client)
+            .map_or(false, |signals| signals.contains(&signal))
+    }
+
+    /**
+     * Check if signal is subscribed to by client
+     */
+    pub fn is_signal_subscribed_by(&self, signal: String, client: String) -> bool {
+        self.signal_to_client
+            .get(&signal)
+            .map_or(false, |clients| clients.contains(&client))
+    }
+}
+
+/**
+ * Unit tests
+ */
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_subscribe_client_to_signal() {
+        let mut subs = Subscriptions::new();
+        subs.subscribe_client_to_signal("client1".to_string(), "signal1".to_string());
+        
+        assert!(subs.is_client_subscribed_to("client1".to_string(), "signal1".to_string()));
+        assert!(subs.is_signal_subscribed_by("signal1".to_string(), "client1".to_string()));
+    }
+
+    #[test]
+    fn test_subscribe_multiple_signals_to_client() {
+        let mut subs = Subscriptions::new();
+        subs.subscribe_client_to_signal("client1".to_string(), "signal1".to_string());
+        subs.subscribe_client_to_signal("client1".to_string(), "signal2".to_string());
+        
+        assert!(subs.is_client_subscribed_to("client1".to_string(), "signal1".to_string()));
+        assert!(subs.is_client_subscribed_to("client1".to_string(), "signal2".to_string()));
+    }
+
+    #[test]
+    fn test_subscribe_multiple_clients_to_signal() {
+        let mut subs = Subscriptions::new();
+        subs.subscribe_client_to_signal("client1".to_string(), "signal1".to_string());
+        subs.subscribe_client_to_signal("client2".to_string(), "signal1".to_string());
+        
+        let clients = subs.get_clients_of_signal("signal1".to_string());
+        assert_eq!(clients.len(), 2);
+        assert!(clients.contains(&&"client1".to_string()));
+        assert!(clients.contains(&&"client2".to_string()));
+    }
+
+    #[test]
+    fn test_unsubscribe_client_from_signal() {
+        let mut subs = Subscriptions::new();
+        subs.subscribe_client_to_signal("client1".to_string(), "signal1".to_string());
+        subs.unsubscribe_client_from_signal("client1".to_string(), "signal1".to_string());
+        
+        assert!(!subs.is_client_subscribed_to("client1".to_string(), "signal1".to_string()));
+        assert!(!subs.is_signal_subscribed_by("signal1".to_string(), "client1".to_string()));
+    }
+
+    #[test]
+    fn test_unsubscribe_removes_empty_entries() {
+        let mut subs = Subscriptions::new();
+        subs.subscribe_client_to_signal("client1".to_string(), "signal1".to_string());
+        subs.unsubscribe_client_from_signal("client1".to_string(), "signal1".to_string());
+        
+        assert!(subs.get_clients_of_signal("signal1".to_string()).is_empty());
+    }
+
+    #[test]
+    fn test_unsubscribe_one_of_multiple_signals() {
+        let mut subs = Subscriptions::new();
+        subs.subscribe_client_to_signal("client1".to_string(), "signal1".to_string());
+        subs.subscribe_client_to_signal("client1".to_string(), "signal2".to_string());
+        subs.unsubscribe_client_from_signal("client1".to_string(), "signal1".to_string());
+        
+        assert!(!subs.is_client_subscribed_to("client1".to_string(), "signal1".to_string()));
+        assert!(subs.is_client_subscribed_to("client1".to_string(), "signal2".to_string()));
+    }
+
+    #[test]
+    fn test_get_clients_of_nonexistent_signal() {
+        let subs = Subscriptions::new();
+        let clients = subs.get_clients_of_signal("nonexistent".to_string());
+        assert!(clients.is_empty());
+    }
+
+    #[test]
+    fn test_double_subscribe_same_client_signal() {
+        let mut subs = Subscriptions::new();
+        subs.subscribe_client_to_signal("client1".to_string(), "signal1".to_string());
+        subs.subscribe_client_to_signal("client1".to_string(), "signal1".to_string());
+        
+        let clients = subs.get_clients_of_signal("signal1".to_string());
+        assert_eq!(clients.len(), 1);
+    }
+}

software/tracksight/rust_backend/src/tasks/api/subscriptions.rs

@@ -0,0 +1,5 @@
+use tokio::sync::broadcast;
+
+pub fn run_api_handler(mut shutdown_rx: broadcast::Receiver<()>) {
+    
+}

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

@@ -1,21 +1,25 @@
 
-use tokio::sync::broadcast;
-use tokio::{select, spawn};
+use tokio::{select};
+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 crate::config::CONFIG;
+use crate::tasks::api::subscriptions::Subscriptions;
 
 use jsoncan_rust::parsing::JsonCanParser;
 use jsoncan_rust::can_database::CanDatabase;
-use jsoncan_rust::can_database; // todo fix the naming conflict with canmessage bruhhh
 
 /**
  * Consumes from serial handler
  * Uses JsonCan config to parse CAN messages and broadcasts to other tasks
  */
-pub async fn run_can_data_handler(mut shutdown_rx: broadcast::Receiver<()>, mut can_queue_rx: broadcast::Receiver<CanPayload>) {
+pub async fn run_can_data_handler(
+    mut shutdown_rx: broadcast::Receiver<()>, 
+    mut can_queue_rx: broadcast::Receiver<CanPayload>,
+    subscriptions: RwLock<Subscriptions>
+) {
     println!("CAN data handler task started.");
 
     let parser = JsonCanParser::new(CONFIG.jsoncan_config_path.clone());
@@ -40,16 +44,24 @@ pub async fn run_can_data_handler(mut shutdown_rx: broadcast::Receiver<()>, mut
                 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;
-                    },
-                };
-
-                for signal in can_message.unpack(can_payload.payload) {
-                    // iterate through subscribers and send signal
+                // 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, 
+                    can_payload.payload, 
+                    Some(can_payload.can_timestamp),
+                );
+
+                for signal in decoded_signals {
+                    for client_id in subscriptions.read().await.get_clients_of_signal(signal.name) {
+                        // todo emit to client
+                    }
                 }
             }
         }

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

@@ -1,4 +1,6 @@
+pub mod api_handler;
 pub mod serial_handler;
 pub mod telem_message;
 pub mod can_data;
-pub mod can_data_handler;
+pub mod can_data_handler;
+pub mod api;

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

@@ -3,7 +3,7 @@ use tokio::select;
 use tokio::sync::{broadcast, mpsc};
 use std::io::{Error, ErrorKind};
 use std::sync::{Arc, atomic::AtomicBool, atomic::Ordering};
-use std::time::Duration;
+use std::time::{Duration, SystemTime};
 use crc::{Crc, CRC_32_ISO_HDLC};
 
 use crate::config::CONFIG;
@@ -163,11 +163,13 @@ fn parse_telem_message(payload: Vec<u8>) -> Result<TelemetryMessage, ()> {
             let can_id = u32::from_le_bytes([payload[1], payload[2], payload[3], payload[4]]);
             // TODO use RTC and NTP time instead of whatever this is
             let can_time_offset = f32::from_le_bytes([payload[5], payload[6], payload[7], payload[8]]);
+            let timestamp = SystemTime::now(); // something with can_time_offset
+
             let can_payload = payload[9..].to_vec();
             TelemetryMessage::Can {
                 body: CanPayload {
                     can_id,
-                    can_time_offset,
+                    can_timestamp: timestamp,
                     payload: can_payload,
                 }
             }