wip: bidirectional websocket

visualizer/Cargo.toml

@@ -13,6 +13,8 @@ itertools = "0.12.0"
 serde = { version = "1.0.164", features = ["derive"] }
 serde_json = "1.0.108"
 url = "2.5.0"
+bevy_tasks = "0.14.0"
+async-channel = "2.5.0"
 
 [dependencies.bevy]
 version = "0.14.0"

visualizer/src/api.rs

@@ -1,10 +1,8 @@
 use bevy::prelude::*;
+use bevy_tasks::AsyncComputeTaskPool;
 use ewebsock::{WsEvent, WsMessage};
 use itertools::Itertools;
-use std::{
-    ops::ControlFlow,
-    sync::{Mutex, mpsc},
-};
+use std::{ops::ControlFlow, sync::Mutex};
 use url::Url;
 
 use crate::Led;
@@ -21,33 +19,50 @@ impl WebsocketPlugin {
 
 impl Plugin for WebsocketPlugin {
     fn build(&self, app: &mut App) {
-        let (sender, receiver) = mpsc::channel();
-        ewebsock::ws_receive(
+        let (inner_sender, ws_receiver) = async_channel::unbounded();
+        let (ws_sender, inner_receiver) = async_channel::unbounded();
+
+        let mut ews_sender = ewebsock::ws_connect(
             self.endpoint.to_string(),
-            Box::new(move |event| match sender.send(event) {
+            Box::new(move |event| match inner_sender.force_send(event) {
                 Ok(_) => ControlFlow::Continue(()),
                 Err(_) => ControlFlow::Break(()),
             }),
         )
         .unwrap();
-        app.insert_resource(Receiver(Mutex::new(receiver)))
+
+        AsyncComputeTaskPool::get().spawn(async move {
+            loop {
+                let event = inner_receiver.recv().await;
+                ews_sender.send(event.unwrap());
+            }
+        });
+
+        app.insert_resource(WsConnection(Mutex::new((ws_sender, ws_receiver))))
             .add_systems(Update, listen_for_frame);
     }
 }
 
-struct Receiver(Mutex<mpsc::Receiver<WsEvent>>);
-impl Resource for Receiver {}
+struct WsConnection(
+    Mutex<(
+        async_channel::Sender<WsMessage>,
+        async_channel::Receiver<WsEvent>,
+    )>,
+);
+impl Resource for WsConnection {}
 
 fn listen_for_frame(
-    recv: Res<Receiver>,
+    recv: Res<WsConnection>,
     query: Query<(&Handle<StandardMaterial>, &Led)>,
     mut materials: ResMut<Assets<StandardMaterial>>,
 ) {
-    let mut last_frame = None;
+    let (sender, receiver) = &mut *recv.0.lock().unwrap();
 
-    while let Ok(event) = recv.0.lock().unwrap().try_recv() {
+    let mut last_frame = None;
+    while let Ok(event) = receiver.try_recv() {
         if let WsEvent::Message(WsMessage::Binary(bytes)) = event {
             last_frame = Some(bytes);
+            break;
         }
     }
 
@@ -61,6 +76,7 @@ fn listen_for_frame(
                 materials.get_mut(material).unwrap().emissive = color.into();
             };
         }
+        let _ = sender.force_send(WsMessage::Binary(vec![1]));
     }
 }
 

webapi/src/visualizer/frame_broadcaster.rs

@@ -77,11 +77,15 @@ impl StreamHandler<lightfx::Frame> for FrameBroadcaster {
 
 pub struct FrameBroadcasterSession {
     server: Addr<FrameBroadcaster>,
+    ready: bool,
 }
 
 impl FrameBroadcasterSession {
     pub fn new(server: Addr<FrameBroadcaster>) -> Self {
-        Self { server }
+        Self {
+            server,
+            ready: false,
+        }
     }
 }
 
@@ -92,9 +96,11 @@ impl Actor for FrameBroadcasterSession {
         self.server.do_send(Connect {
             addr: ctx.address().recipient(),
         });
+        self.ready = true;
     }
 
     fn stopped(&mut self, ctx: &mut Self::Context) {
+        self.ready = false;
         self.server.do_send(Disconnect {
             addr: ctx.address().recipient(),
         })
@@ -105,15 +111,28 @@ impl Handler<Frame> for FrameBroadcasterSession {
     type Result = ();
 
     fn handle(&mut self, msg: Frame, ctx: &mut Self::Context) -> Self::Result {
+        if !self.ready {
+            return;
+        }
+
         let bytes: Vec<_> = msg
             .frame
             .pixels_iter()
             .flat_map(|c| [c.r, c.g, c.b])
             .collect();
         ctx.binary(bytes);
+        self.ready = false;
     }
 }
 
 impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for FrameBroadcasterSession {
-    fn handle(&mut self, _item: Result<ws::Message, ws::ProtocolError>, _ctx: &mut Self::Context) {}
+    fn handle(&mut self, item: Result<ws::Message, ws::ProtocolError>, _ctx: &mut Self::Context) {
+        let Ok(ws::Message::Binary(bytes)) = item else {
+            return;
+        };
+
+        if bytes[..] == [1] {
+            self.ready = true;
+        }
+    }
 }