eculocate

commit 71460edfdb652e3a0b4e1159e3560da0956b4ec8
parent de2b0cef90a4dfe17e55e2d41e52da3bb2ff0f31
Author: Daniel Barlow <dan@telent.net>
Date:   Wed,  7 Jan 2026 23:50:35 +0000

rearrange subscriber and subscription

Diffstat:
M
ecu-esp32/src/streamer.rs
 | 
105
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------------------

1 file changed, 79 insertions(+), 26 deletions(-)
diff --git a/ecu-esp32/src/streamer.rs b/ecu-esp32/src/streamer.rs
@@ -4,24 +4,66 @@ use embassy_net::{
     Stack
 };
 use embassy_time::{Duration, Instant, Timer};
+use embassy_futures::select::select;
 
+use alloc::vec::Vec;
 
 use log::info;
+
 use crate::ecu::Ecu;
 
-#[derive(Copy, Clone)]
+#[derive(Debug)]
 enum SubscriptionKind {
     Table(u8, usize, usize)
 }
 
-#[derive(Copy, Clone)]
+#[derive(Debug)]
 struct Subscription {
-    kind : SubscriptionKind,
+    kind: SubscriptionKind,
     interval: Duration,
-    subscriber: IpEndpoint,
+    next_wake: Instant
+}
+impl Subscription {
+    fn new(kind: SubscriptionKind, interval: Duration) -> Self {
+        Self {
+            kind,
+            interval,
+            next_wake: Instant::now() + interval
+        }
+    }
+}
+
+struct Subscriber {
+    subscriptions: Vec<Subscription>,
+    endpoint: IpEndpoint,
     until: Instant
 }
 
+impl Subscriber {
+    fn next_wake(&self) -> Option<Instant> {
+        self.subscriptions.iter().map(|s| s.next_wake).min()
+    }
+
+    fn update_subscriptions(&mut self, message: &[u8]) {
+        let mut subs = Vec::new();
+        for c in message.chunks(4) {
+            let interval : u64 = (u64::from(c[0]) << 8)  + u64::from(c[1]);
+            let sub_type = c[2];
+            let table_number = c[3];
+            if interval > 0 && sub_type == b'T' {
+                subs.push(Subscription::new(SubscriptionKind::Table(table_number, 0, 40),
+                                            Duration::from_millis(interval)))
+            }
+        }
+        info!("subs {:?}", subs);
+        self.subscriptions = subs;
+    }
+}
+
+// endpoint and timeout are per-subscriber not per subscription.
+// interval is per subscription: e.g. we want to get the rpm much
+// more often than the ecm id
+
 pub async fn loop_udp_thing(stack : Stack<'_>, ecu: &mut Ecu<'_>) {
     let mut rx_buffer = [0u8; 4096];
     let mut tx_buffer = [0u8; 4096];
@@ -38,39 +80,50 @@ pub async fn loop_udp_thing(stack : Stack<'_>, ecu: &mut Ecu<'_>) {
 
     log::info!("sock.bind 5000 res: {:?}", sock.bind(5000));
 
-    let mut subscription = None;
+    // TODO we need a map(dictionary) of endpoint -> subscriber
+
+    let mut subscriber = None;
 
     loop {
         let mut buf = [0; 4096];
         if sock.may_recv() {
             info!("packet");
             let res = sock.recv_from(&mut buf).await;
-            if let Ok((_n, meta)) = res {
+            if let Ok((n, meta)) = res {
                 info!("packet from {:?}", meta.endpoint);
-                // TODO don't hardcode, parse the actual message to find
-                // out what they wanted to subscribe to
-                subscription = Some(Subscription {
-                    kind: SubscriptionKind::Table(0x11, 0, 25),
-                    interval: Duration::from_millis(500),
-                    subscriber: meta.endpoint,
-		    until: Instant::now() + Duration::from_secs(60),
-                })
+
+                let mut s = Subscriber {
+                    subscriptions: Vec::from([]),
+                    endpoint: meta.endpoint,
+                    until: Instant::now() + Duration::from_secs(60),
+                };
+                s.update_subscriptions(&buf[0..n]);
+                subscriber = Some(s);
             }
         }
-        if let Some(sub) = subscription {
+        if let Some(ref mut sub) = subscriber {
             let mut buf = [0u8; 1024];
-	    if sub.until < Instant::now() {
-		subscription = None;
-	    } else {
-		match sub.kind {
-                    SubscriptionKind::Table(t, _s, _e) => {
-			let tbl = ecu.fetch_table(t, &mut buf).await;
-			let p = sock.send_to(tbl, sub.subscriber).await;
-			info!("sent {:?}", p);
+            let now = Instant::now();
+            if sub.until < now {
+                subscriber = None;
+            } else {
+                let subn = &mut sub.subscriptions[0];
+                if subn.next_wake <= now {
+                    match subn.kind {
+                        SubscriptionKind::Table(t, _s, _e) => {
+                            let tbl = ecu.fetch_table(t, &mut buf).await;
+                        let p = sock.send_to(tbl, sub.endpoint).await;
+                            info!("sent {:?}", p);
+                        }
                     }
-		}
-		Timer::after(sub.interval).await;
-	    }
+                    subn.next_wake = now + subn.interval;
+                }
+                if let Some(later) = sub.next_wake() {
+                    select(Timer::at(later), sock.wait_recv_ready()).await;
+                } else {
+                    sock.wait_recv_ready().await;
+                }
+            }
         } else {
             sock.wait_recv_ready().await;
         }