Making subscription working with fake data generator, no can communication at this point

Author: erdemsimsek

Cargo.lock

@@ -10,3 +10,5 @@ rfd = "0.15.4"
 egui_plot = "0.31.0"
 image = "0.25.1"
 chrono = "0.4.41"
+tokio = { version = "1.47.1", features = ["full"] }
+rand = "0.9.2"

Cargo.toml

@@ -1,7 +1,9 @@
 use std::sync::mpsc::{Receiver, Sender};
 use std::path::PathBuf;
 use configparser::ini::Ini;
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
+use rand::Rng;
+use tokio::{sync::{mpsc as tokio_mpsc, oneshot}, task::JoinHandle};
 
 
 #[derive(Debug, Clone)]
@@ -47,6 +49,22 @@ pub enum Update {
     },
 }
 
+async fn simulation_task(address: SdoAddress, interval_ms: u64, update_tx: Sender<Update>) {
+    println!("Starting simulation task for address {:?} with interval {} ms", &address, interval_ms);
+    let mut interval = tokio::time::interval(std::time::Duration::from_millis(interval_ms));
+
+
+    loop {
+        interval.tick().await;
+        let mut rng = rand::thread_rng();
+        let random_value = rng.gen_range(0..100);
+        let _ = update_tx.send(Update::SdoData {
+            address: address.clone(),
+            value: format!("{}", random_value),
+        });
+    }
+}
+
 // Make the main function for the thread public as well.
 pub fn communication_thread_main(
     command_rx: Receiver<Command>,
@@ -56,9 +74,12 @@ pub fn communication_thread_main(
     eds_file: Option<PathBuf>,
 ) {
 
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    let mut subscription_handles: HashMap<SdoAddress, JoinHandle<()>> = HashMap::new();
 
     for command in command_rx {
         match command {
+            Command::Connect => {},
             Command::FetchSdos => {
                 if let Some(path) = eds_file.as_ref() {
                     match search_for_readable_sdo(path.clone()) {
@@ -74,7 +95,19 @@ pub fn communication_thread_main(
                     let _ = update_tx.send(Update::SdoList(BTreeMap::new()));
                 }
             },
-            Command::Connect => {},
+            Command::Subscribe { address, interval_ms } => {
+                println!("Subscribing to address {:?} with interval {} ms", &address, interval_ms);
+                let update_tx_clone = update_tx.clone();
+                let subscription_handle = rt.spawn(simulation_task(address.clone(), interval_ms, update_tx_clone));
+                subscription_handles.insert(address, subscription_handle);
+            },
+            Command::Unsubscribe(address) => {
+                println!("Unsubscribing from address {:?}", &address);
+                let subscription_handle = subscription_handles.remove(&address);
+                if let Some(subscription_handle) = subscription_handle {
+                    subscription_handle.abort();
+                }
+            }
             _ => {}
         }
     }

src/communication.rs

@@ -2,7 +2,7 @@
 
 mod communication;
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, VecDeque};
 use std::ops::Deref;
 use communication::{Command, Update, SdoAddress, SdoObject};
 
@@ -14,6 +14,8 @@ use egui_plot::{Plot, PlotPoints, Line};
 use chrono::Local;
 use std::sync::Arc;
 
+const PLOT_BUFFER_SIZE: usize = 500;
+
 enum AppView {
     SelectInterface,
     SelectNodeId,
@@ -24,7 +26,7 @@ enum AppView {
 #[derive(Debug, Clone)]
 struct SdoSubscription{
     interval_ms: u64,
-    plot_data: Vec<[f64; 2]>,
+    plot_data: VecDeque<[f64; 2]>,
 }
 struct ScreenshotInfo {
     filename: String,
@@ -98,6 +100,21 @@ impl eframe::App for MyApp {
             match update{
                 Update::SdoList(map) => {
                     self.sdo_data = Some(map);
+                },
+
+                Update::SdoData { address, value } => {
+                    // 1. Try to parse the incoming string value into a number.
+                    if let Ok(number_value) = value.parse::<f64>() {
+                        // 2. Find the subscription this data belongs to.
+                        if let Some(subscription) = self.subscriptions.get_mut(&address) {
+
+                            if subscription.plot_data.len() >= PLOT_BUFFER_SIZE {
+                                subscription.plot_data.pop_front();
+                            }
+                            let time = subscription.plot_data.back().map_or(0.0, |p| p[0] + 1.0);
+                            subscription.plot_data.push_back([time, number_value]);
+                        }
+                    }
                 }
                 _ => {
 
@@ -125,6 +142,8 @@ impl eframe::App for MyApp {
                 AppView::Main => self.draw_main_view(ui),
             }
         });
+
+        ctx.request_repaint();
     }
 }
 
@@ -376,6 +395,8 @@ impl MyApp {
                             .allow_scroll(false)
                             .height(250.0)
                             .width(ui.available_width())
+                            .x_axis_label("Sample No")
+                            .y_axis_label("Value")
                             .show(ui, |plot_ui| {
                                 // 2. Generate a unique color for the line based on its address.
                                 let color = Color32::from_rgb(
@@ -384,10 +405,12 @@ impl MyApp {
                                     (address.index as u8 ^ address.sub_index as u8).wrapping_mul(30),
                                 );
 
-                                let line = Line::new(PlotPoints::from(subscription.plot_data.clone()))
-                                    .name(&plot_title) // Use the title for the legend as well
-                                    .color(color);
+                                let points_vec: Vec<[f64; 2]> = subscription.plot_data.iter().cloned().collect();
 
+                                let line = Line::new(PlotPoints::from(points_vec))
+                                    .name(&plot_title)
+                                    .color(color);
+                                
                                 plot_ui.line(line);
                             });
 
@@ -439,7 +462,7 @@ impl MyApp {
                                 }
                                 self.subscriptions.insert(address.clone(), SdoSubscription {
                                     interval_ms,
-                                    plot_data: Vec::new(),
+                                    plot_data: VecDeque::new(),
                                 });
                                 self.modal_open_for = None; // Close the modal
                             }