open udp socket, write table 0x11 to any peer that sends us data

TODO: actually parse the message from the peer in case they want
somethng other than table 0x11

ecu-esp32/src/main.rs

@@ -12,6 +12,7 @@ use esp_println as _;
 pub mod ecu;
 pub mod wifi;
 pub mod mdns;
+pub mod streamer;
 
 use esp_hal::{
     timer::timg::TimerGroup,
@@ -64,20 +65,20 @@ async fn main(s : Spawner) {
     // wibble::run_ble(peripherals.BT, ecu).await;
 
     join(
-	crate::wifi::start_wifi(&s, peripherals.WIFI),
+	crate::wifi::start_wifi(&s, peripherals.WIFI, &mut ecu),
 	async {
 	    loop {
-		let c = async |measure| {
-		    match measure {
-			Measure::Rpm(v) => {
-			    info!("got rpm {v}");
-			},
-			Measure::Tps(mv, percentage) => {
-			    info!("got tps {mv} {percentage}");
-			}
-		    }
-		};
-		let _ = &ecu.poll(c).await;
+		// let c = async |measure| {
+		//     match measure {
+		// 	Measure::Rpm(v) => {
+		// 	    info!("got rpm {v}");
+		// 	},
+		// 	Measure::Tps(mv, percentage) => {
+		// 	    info!("got tps {mv} {percentage}");
+		// 	}
+		//     }
+		// };
+		// let _ = &ecu.poll(c).await;
 		Timer::after_millis(1000).await;
 	    }
     }).await;

ecu-esp32/src/streamer.rs

@@ -0,0 +1,73 @@
+use embassy_net::{
+    udp::{UdpSocket, PacketMetadata},
+    IpEndpoint,
+    Stack
+};
+use embassy_time::{Duration, Timer};
+
+
+use log::info;
+use crate::ecu::Ecu;
+
+#[derive(Copy, Clone)]
+enum SubscriptionKind {
+    Table(u8, usize, usize)
+}
+
+#[derive(Copy, Clone)]
+struct Subscription {
+    kind : SubscriptionKind,
+    interval: Duration,
+    subscriber: IpEndpoint
+}
+
+pub async fn loop_udp_thing(stack : Stack<'_>, ecu: &mut Ecu<'_>) {
+    let mut rx_buffer = [0u8; 4096];
+    let mut tx_buffer = [0u8; 4096];
+    let mut rx_meta = [PacketMetadata::EMPTY; 512];
+    let mut tx_meta = [PacketMetadata::EMPTY; 512];
+
+    let mut sock = UdpSocket::new(
+        stack,
+        &mut rx_meta,
+        &mut rx_buffer,
+        &mut tx_meta,
+        &mut tx_buffer,
+    );
+
+    log::info!("sock.bind 5000 res: {:?}", sock.bind(5000));
+
+    let mut subscription = 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 {
+                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
+                })
+            }
+        }
+        if let Some(sub) = subscription {
+            info!("got subscriber");
+            let mut buf = [0u8; 1024];
+            match sub.kind {
+                SubscriptionKind::Table(t, _s, _e) => {
+                    let tbl = ecu.fetch_table(t, &mut buf).await;
+                    info!("sending {}", tbl.len());
+                    let _ = sock.send_to(tbl, sub.subscriber).await;
+                }
+            }
+            Timer::after(sub.interval).await;
+        } else {
+            sock.wait_recv_ready().await;
+        }
+    }
+}

ecu-esp32/src/wifi.rs

@@ -12,6 +12,7 @@ use embassy_net::{
     Runner,
     StackResources
 };
+use embassy_futures::join::join;
 
 use embassy_time::{Duration, Timer};
 use esp_alloc as _;
@@ -85,7 +86,8 @@ fn dns_records(my_ip_address: [u8;4], serial: Option<usize>) -> Vec<mdns::Resour
 }
 
 pub async fn start_wifi(spawner: &Spawner,
-		    wifi_peripheral:  esp_hal::peripherals::WIFI<'static>) -> ! {
+		        wifi_peripheral:  esp_hal::peripherals::WIFI<'static>,
+                        ecu: &mut crate::ecu::Ecu<'_>)  {
 
 
     let esp_radio_ctrl = &*mk_static!(Controller<'static>, esp_radio::init().unwrap());
@@ -131,18 +133,24 @@ pub async fn start_wifi(spawner: &Spawner,
     }
 
     let mut serial = None;
-    loop {
-	let mdns_thread = mdns::responder(
-	    stack,
-	    dns_records(my_address, serial)
-	);
-	mdns_thread.await;
-	serial = match serial { None => Some(2), Some(n) => Some(n+1) };
-	info!("probe failed, restarting with incremented serial {:?}", serial);
-        Timer::after(Duration::from_millis(5000)).await;
-    };
+    join(
+        crate::streamer::loop_udp_thing(stack, ecu),
+        async {
+            loop {
+	        let mdns_thread = mdns::responder(
+	            stack,
+	            dns_records(my_address, serial)
+	        );
+	        mdns_thread.await;
+	        serial = match serial { None => Some(2), Some(n) => Some(n+1) };
+	        info!("probe failed, restarting with incremented serial {:?}", serial);
+                Timer::after(Duration::from_millis(5000)).await;
+            };
+        }
+    ).await;
 }
 
+
 #[embassy_executor::task]
 async fn connection(mut controller: WifiController<'static>) {
     println!("start connection task");