chore: simplify liveview abstraction

packages/liveview/src/adapters/warp_adapter.rs

@@ -1,12 +1,14 @@
-use crate::{LiveView, LiveViewError};
+use crate::{liveview_eventloop, LiveView, LiveViewError};
 use dioxus_core::prelude::*;
-use dioxus_html::HtmlEvent;
 use futures_util::{SinkExt, StreamExt};
-use std::time::Duration;
 use warp::ws::{Message, WebSocket};
 
 impl LiveView {
-    pub async fn upgrade_warp(self, ws: WebSocket, app: fn(Scope<()>) -> Element) {
+    pub async fn upgrade_warp(
+        self,
+        ws: WebSocket,
+        app: fn(Scope<()>) -> Element,
+    ) -> Result<(), LiveViewError> {
         self.upgrade_warp_with_props(ws, app, ()).await
     }
 
@@ -15,60 +17,37 @@ impl LiveView {
         ws: WebSocket,
         app: fn(Scope<T>) -> Element,
         props: T,
-    ) {
-        self.pool
-            .spawn_pinned(move || liveview_eventloop(app, props, ws))
-            .await;
-    }
-}
+    ) -> Result<(), LiveViewError> {
+        let (ws_tx, ws_rx) = ws.split();
 
-async fn liveview_eventloop<T>(
-    app: Component<T>,
-    props: T,
-    mut ws: WebSocket,
-) -> Result<(), LiveViewError>
-where
-    T: Send + 'static,
-{
-    let mut vdom = VirtualDom::new_with_props(app, props);
-    let edits = serde_json::to_string(&vdom.rebuild()).unwrap();
-    ws.send(Message::text(edits)).await.unwrap();
+        let ws_tx = ws_tx
+            .with(transform_warp)
+            .sink_map_err(|_| LiveViewError::SendingFailed);
 
-    loop {
-        tokio::select! {
-            // poll any futures or suspense
-            _ = vdom.wait_for_work() => {}
+        let ws_rx = ws_rx.map(transform_warp_rx);
 
-            evt = ws.next() => {
-                match evt {
-                    Some(Ok(evt)) => {
-                        if let Ok(evt) = evt.to_str() {
-                            // desktop uses this wrapper struct thing
-                            #[derive(serde::Deserialize)]
-                            struct IpcMessage {
-                                params: HtmlEvent
-                            }
-
-                            let event: IpcMessage = serde_json::from_str(evt).unwrap();
-                            let event = event.params;
-                            let bubbles = event.bubbles();
-                            vdom.handle_event(&event.name, event.data.into_any(), event.element, bubbles);
-                        }
-                    }
-                    Some(Err(_e)) => {
-                        // log this I guess?
-                        // when would we get an error here?
-                    }
-                    None => return Ok(()),
-                }
-            }
+        match self
+            .pool
+            .spawn_pinned(move || liveview_eventloop(app, props, ws_tx, ws_rx))
+            .await
+        {
+            Ok(Ok(_)) => Ok(()),
+            Ok(Err(e)) => Err(e),
+            Err(_) => Err(LiveViewError::SendingFailed),
         }
-
-        let edits = vdom
-            .render_with_deadline(tokio::time::sleep(Duration::from_millis(10)))
-            .await;
-
-        ws.send(Message::text(serde_json::to_string(&edits).unwrap()))
-            .await?;
     }
 }
+
+fn transform_warp_rx(f: Result<Message, warp::Error>) -> Result<String, LiveViewError> {
+    // destructure the message into the buffer we got from warp
+    let msg = f.map_err(|_| LiveViewError::SendingFailed)?.into_bytes();
+
+    // transform it back into a string, saving us the allocation
+    let msg = String::from_utf8(msg).map_err(|_| LiveViewError::SendingFailed)?;
+
+    Ok(msg)
+}
+
+async fn transform_warp(message: String) -> Result<Message, warp::Error> {
+    Ok(Message::text(message))
+}

packages/liveview/src/lib.rs

@@ -1,3 +1,30 @@
+pub mod adapters {
+    #[cfg(feature = "warp")]
+    pub mod warp_adapter;
+
+    #[cfg(feature = "axum")]
+    pub mod axum_adapter;
+
+    #[cfg(feature = "salvo")]
+    pub mod salvo_adapter;
+}
+
+pub mod pool;
+use futures_util::{SinkExt, StreamExt};
+pub use pool::*;
+
+pub trait WebsocketTx: SinkExt<String, Error = LiveViewError> {}
+impl<T> WebsocketTx for T where T: SinkExt<String, Error = LiveViewError> {}
+
+pub trait WebsocketRx: StreamExt<Item = Result<String, LiveViewError>> {}
+impl<T> WebsocketRx for T where T: StreamExt<Item = Result<String, LiveViewError>> {}
+
+#[derive(Debug, thiserror::Error)]
+pub enum LiveViewError {
+    #[error("warp error")]
+    SendingFailed,
+}
+
 use dioxus_interpreter_js::INTERPRETER_JS;
 static MAIN_JS: &str = include_str!("./main.js");
 
@@ -13,23 +40,3 @@ pub fn interpreter_glue(url: &str) -> String {
     "#
     )
 }
-
-pub mod adapters {
-    #[cfg(feature = "warp")]
-    pub mod warp_adapter;
-
-    #[cfg(feature = "axum")]
-    pub mod axum_adapter;
-
-    #[cfg(feature = "salvo")]
-    pub mod salvo_adapter;
-}
-
-pub mod pool;
-pub use pool::*;
-
-#[derive(Debug, thiserror::Error)]
-pub enum LiveViewError {
-    #[error("Connection Failed")]
-    Warp(#[from] warp::Error),
-}

packages/liveview/src/main.js

@@ -4,7 +4,7 @@ function main() {
   if (root != null) {
     // create a new ipc
     window.ipc = new IPC(root);
-    window.ipc.send(serializeIpcMessage("initialize"));
+    window.ipc.postMessage(serializeIpcMessage("initialize"));
   }
 }
 
@@ -24,6 +24,7 @@ class IPC {
     };
 
     this.ws.onmessage = (event) => {
+      console.log("Received message: ", event.data);
       let edits = JSON.parse(event.data);
       window.interpreter.handleEdits(edits);
     };

packages/liveview/src/pool.rs

@@ -1,3 +1,8 @@
+use crate::LiveViewError;
+use dioxus_core::prelude::*;
+use dioxus_html::HtmlEvent;
+use futures_util::{pin_mut, SinkExt, StreamExt};
+use std::time::Duration;
 use tokio_util::task::LocalPoolHandle;
 
 #[derive(Clone)]
@@ -18,3 +23,66 @@ impl LiveView {
         }
     }
 }
+
+/// The primary event loop for the VirtualDom waiting for user input
+///
+/// This function makes it easy to integrate Dioxus LiveView with any socket-based framework.
+///
+/// As long as your framework can provide a Sink and Stream of Strings, you can use this function.
+///
+/// You might need to transform the error types of the web backend into the LiveView error type.
+pub async fn liveview_eventloop<T>(
+    app: Component<T>,
+    props: T,
+    ws_tx: impl SinkExt<String, Error = LiveViewError>,
+    ws_rx: impl StreamExt<Item = Result<String, LiveViewError>>,
+) -> Result<(), LiveViewError>
+where
+    T: Send + 'static,
+{
+    let mut vdom = VirtualDom::new_with_props(app, props);
+
+    // todo: use an efficient binary packed format for this
+    let edits = serde_json::to_string(&vdom.rebuild()).unwrap();
+
+    // pin the futures so we can use select!
+    pin_mut!(ws_tx);
+    pin_mut!(ws_rx);
+
+    ws_tx.send(edits).await?;
+
+    // desktop uses this wrapper struct thing around the actual event itself
+    // this is sorta driven by tao/wry
+    #[derive(serde::Deserialize)]
+    struct IpcMessage {
+        params: HtmlEvent,
+    }
+
+    loop {
+        tokio::select! {
+            // poll any futures or suspense
+            _ = vdom.wait_for_work() => {}
+
+            evt = ws_rx.next() => {
+                match evt {
+                    Some(Ok(evt)) => {
+                        let event: IpcMessage = serde_json::from_str(&evt).unwrap();
+                        let event = event.params;
+                        vdom.handle_event(&event.name, event.data.into_any(), event.element, event.bubbles);
+                    }
+                    Some(Err(_e)) => {
+                        // log this I guess?
+                        // when would we get an error here?
+                    }
+                    None => return Ok(()),
+                }
+            }
+        }
+
+        let edits = vdom
+            .render_with_deadline(tokio::time::sleep(Duration::from_millis(10)))
+            .await;
+
+        ws_tx.send(serde_json::to_string(&edits).unwrap()).await?;
+    }
+}