Revert "fix: liveview interpreter using new templates"

packages/liveview/src/adapters/axum_adapter.rs

@@ -1,23 +1,94 @@
-use crate::{LiveViewError, LiveViewSocket};
+use crate::events;
 use axum::extract::ws::{Message, WebSocket};
-use futures_util::{SinkExt, StreamExt};
+use dioxus_core::prelude::*;
+use futures_util::{
+    future::{select, Either},
+    pin_mut, SinkExt, StreamExt,
+};
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::UnboundedReceiverStream;
+use tokio_util::task::LocalPoolHandle;
 
-/// Convert a warp websocket into a LiveViewSocket
-///
-/// This is required to launch a LiveView app using the warp web framework
-pub fn axum_socket(ws: WebSocket) -> impl LiveViewSocket {
-    ws.map(transform_rx)
-        .with(transform_tx)
-        .sink_map_err(|_| LiveViewError::SendingFailed)
+impl crate::Liveview {
+    pub async fn upgrade_axum(&self, ws: WebSocket, app: fn(Scope) -> Element) {
+        connect(ws, self.pool.clone(), app, ()).await;
+    }
+
+    pub async fn upgrade_axum_with_props<T>(
+        &self,
+        ws: WebSocket,
+        app: fn(Scope<T>) -> Element,
+        props: T,
+    ) where
+        T: Send + Sync + 'static,
+    {
+        connect(ws, self.pool.clone(), app, props).await;
+    }
 }
 
-fn transform_rx(message: Result<Message, axum::Error>) -> Result<String, LiveViewError> {
-    message
-        .map_err(|_| LiveViewError::SendingFailed)?
-        .into_text()
-        .map_err(|_| LiveViewError::SendingFailed)
-}
-
-async fn transform_tx(message: String) -> Result<Message, axum::Error> {
-    Ok(Message::Text(message))
+pub async fn connect<T>(
+    socket: WebSocket,
+    pool: LocalPoolHandle,
+    app: fn(Scope<T>) -> Element,
+    props: T,
+) where
+    T: Send + Sync + 'static,
+{
+    let (mut user_ws_tx, mut user_ws_rx) = socket.split();
+    let (event_tx, event_rx) = mpsc::unbounded_channel();
+    let (edits_tx, edits_rx) = mpsc::unbounded_channel();
+    let mut edits_rx = UnboundedReceiverStream::new(edits_rx);
+    let mut event_rx = UnboundedReceiverStream::new(event_rx);
+    let vdom_fut = pool.clone().spawn_pinned(move || async move {
+        let mut vdom = VirtualDom::new_with_props(app, props);
+        let edits = vdom.rebuild();
+        let serialized = serde_json::to_string(&edits.edits).unwrap();
+        edits_tx.send(serialized).unwrap();
+        loop {
+            let new_event = {
+                let vdom_fut = vdom.wait_for_work();
+                pin_mut!(vdom_fut);
+                match select(event_rx.next(), vdom_fut).await {
+                    Either::Left((l, _)) => l,
+                    Either::Right((_, _)) => None,
+                }
+            };
+            if let Some(new_event) = new_event {
+                vdom.handle_message(dioxus_core::SchedulerMsg::Event(new_event));
+            } else {
+                let mutations = vdom.work_with_deadline(|| false);
+                for mutation in mutations {
+                    let edits = serde_json::to_string(&mutation.edits).unwrap();
+                    edits_tx.send(edits).unwrap();
+                }
+            }
+        }
+    });
+    loop {
+        match select(user_ws_rx.next(), edits_rx.next()).await {
+            Either::Left((l, _)) => {
+                if let Some(Ok(msg)) = l {
+                    if let Ok(Some(msg)) = msg.to_text().map(events::parse_ipc_message) {
+                        let user_event = events::trigger_from_serialized(msg.params);
+                        event_tx.send(user_event).unwrap();
+                    } else {
+                        break;
+                    }
+                } else {
+                    break;
+                }
+            }
+            Either::Right((edits, _)) => {
+                if let Some(edits) = edits {
+                    // send the edits to the client
+                    if user_ws_tx.send(Message::Text(edits)).await.is_err() {
+                        break;
+                    }
+                } else {
+                    break;
+                }
+            }
+        }
+    }
+    vdom_fut.abort();
 }

packages/liveview/src/adapters/salvo_adapter.rs

@@ -1,25 +1,110 @@
-use futures_util::{SinkExt, StreamExt};
+use crate::events;
+use dioxus_core::prelude::*;
+use futures_util::{pin_mut, SinkExt, StreamExt};
 use salvo::extra::ws::{Message, WebSocket};
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::UnboundedReceiverStream;
+use tokio_util::task::LocalPoolHandle;
 
-use crate::{LiveViewError, LiveViewSocket};
-
-/// Convert a salvo websocket into a LiveViewSocket
-///
-/// This is required to launch a LiveView app using the warp web framework
-pub fn salvo_socket(ws: WebSocket) -> impl LiveViewSocket {
-    ws.map(transform_rx)
-        .with(transform_tx)
-        .sink_map_err(|_| LiveViewError::SendingFailed)
+impl crate::Liveview {
+    pub async fn upgrade_salvo(&self, ws: salvo::extra::ws::WebSocket, app: fn(Scope) -> Element) {
+        connect(ws, self.pool.clone(), app, ()).await;
+    }
+    pub async fn upgrade_salvo_with_props<T>(
+        &self,
+        ws: salvo::extra::ws::WebSocket,
+        app: fn(Scope<T>) -> Element,
+        props: T,
+    ) where
+        T: Send + Sync + 'static,
+    {
+        connect(ws, self.pool.clone(), app, props).await;
+    }
 }
 
-fn transform_rx(message: Result<Message, salvo::Error>) -> Result<String, LiveViewError> {
-    let as_bytes = message.map_err(|_| LiveViewError::SendingFailed)?;
+pub async fn connect<T>(
+    ws: WebSocket,
+    pool: LocalPoolHandle,
+    app: fn(Scope<T>) -> Element,
+    props: T,
+) where
+    T: Send + Sync + 'static,
+{
+    // Use a counter to assign a new unique ID for this user.
 
-    let msg = String::from_utf8(as_bytes.into_bytes()).map_err(|_| LiveViewError::SendingFailed)?;
+    // Split the socket into a sender and receive of messages.
+    let (mut user_ws_tx, mut user_ws_rx) = ws.split();
 
-    Ok(msg)
-}
+    let (event_tx, event_rx) = mpsc::unbounded_channel();
+    let (edits_tx, edits_rx) = mpsc::unbounded_channel();
 
-async fn transform_tx(message: String) -> Result<Message, salvo::Error> {
-    Ok(Message::text(message))
+    let mut edits_rx = UnboundedReceiverStream::new(edits_rx);
+    let mut event_rx = UnboundedReceiverStream::new(event_rx);
+
+    let vdom_fut = pool.spawn_pinned(move || async move {
+        let mut vdom = VirtualDom::new_with_props(app, props);
+
+        let edits = vdom.rebuild();
+
+        let serialized = serde_json::to_string(&edits.edits).unwrap();
+        edits_tx.send(serialized).unwrap();
+
+        loop {
+            use futures_util::future::{select, Either};
+
+            let new_event = {
+                let vdom_fut = vdom.wait_for_work();
+
+                pin_mut!(vdom_fut);
+
+                match select(event_rx.next(), vdom_fut).await {
+                    Either::Left((l, _)) => l,
+                    Either::Right((_, _)) => None,
+                }
+            };
+
+            if let Some(new_event) = new_event {
+                vdom.handle_message(dioxus_core::SchedulerMsg::Event(new_event));
+            } else {
+                let mutations = vdom.work_with_deadline(|| false);
+                for mutation in mutations {
+                    let edits = serde_json::to_string(&mutation.edits).unwrap();
+                    edits_tx.send(edits).unwrap();
+                }
+            }
+        }
+    });
+
+    loop {
+        use futures_util::future::{select, Either};
+
+        match select(user_ws_rx.next(), edits_rx.next()).await {
+            Either::Left((l, _)) => {
+                if let Some(Ok(msg)) = l {
+                    if let Ok(Some(msg)) = msg.to_str().map(events::parse_ipc_message) {
+                        if msg.method == "user_event" {
+                            let user_event = events::trigger_from_serialized(msg.params);
+                            event_tx.send(user_event).unwrap();
+                        }
+                    } else {
+                        break;
+                    }
+                } else {
+                    break;
+                }
+            }
+            Either::Right((edits, _)) => {
+                if let Some(edits) = edits {
+                    // send the edits to the client
+                    if user_ws_tx.send(Message::text(edits)).await.is_err() {
+                        break;
+                    }
+                } else {
+                    break;
+                }
+            }
+        }
+    }
+
+    vdom_fut.abort();
 }

packages/liveview/src/adapters/warp_adapter.rs

@@ -1,28 +1,110 @@
-use crate::{LiveViewError, LiveViewSocket};
-use futures_util::{SinkExt, StreamExt};
+use crate::events;
+use dioxus_core::prelude::*;
+use futures_util::{pin_mut, SinkExt, StreamExt};
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::UnboundedReceiverStream;
+use tokio_util::task::LocalPoolHandle;
 use warp::ws::{Message, WebSocket};
 
-/// Convert a warp websocket into a LiveViewSocket
-///
-/// This is required to launch a LiveView app using the warp web framework
-pub fn warp_socket(ws: WebSocket) -> impl LiveViewSocket {
-    ws.map(transform_rx)
-        .with(transform_tx)
-        .sink_map_err(|_| LiveViewError::SendingFailed)
+impl crate::Liveview {
+    pub async fn upgrade_warp(&self, ws: warp::ws::WebSocket, app: fn(Scope) -> Element) {
+        connect(ws, self.pool.clone(), app, ()).await;
+    }
+    pub async fn upgrade_warp_with_props<T>(
+        &self,
+        ws: warp::ws::WebSocket,
+        app: fn(Scope<T>) -> Element,
+        props: T,
+    ) where
+        T: Send + Sync + 'static,
+    {
+        connect(ws, self.pool.clone(), app, props).await;
+    }
 }
 
-fn transform_rx(message: Result<Message, warp::Error>) -> Result<String, LiveViewError> {
-    // destructure the message into the buffer we got from warp
-    let msg = message
-        .map_err(|_| LiveViewError::SendingFailed)?
-        .into_bytes();
+pub async fn connect<T>(
+    ws: WebSocket,
+    pool: LocalPoolHandle,
+    app: fn(Scope<T>) -> Element,
+    props: T,
+) where
+    T: Send + Sync + 'static,
+{
+    // Use a counter to assign a new unique ID for this user.
 
-    // transform it back into a string, saving us the allocation
-    let msg = String::from_utf8(msg).map_err(|_| LiveViewError::SendingFailed)?;
+    // Split the socket into a sender and receive of messages.
+    let (mut user_ws_tx, mut user_ws_rx) = ws.split();
 
-    Ok(msg)
-}
+    let (event_tx, event_rx) = mpsc::unbounded_channel();
+    let (edits_tx, edits_rx) = mpsc::unbounded_channel();
 
-async fn transform_tx(message: String) -> Result<Message, warp::Error> {
-    Ok(Message::text(message))
+    let mut edits_rx = UnboundedReceiverStream::new(edits_rx);
+    let mut event_rx = UnboundedReceiverStream::new(event_rx);
+
+    let vdom_fut = pool.spawn_pinned(move || async move {
+        let mut vdom = VirtualDom::new_with_props(app, props);
+
+        let edits = vdom.rebuild();
+
+        let serialized = serde_json::to_string(&edits.edits).unwrap();
+        edits_tx.send(serialized).unwrap();
+
+        loop {
+            use futures_util::future::{select, Either};
+
+            let new_event = {
+                let vdom_fut = vdom.wait_for_work();
+
+                pin_mut!(vdom_fut);
+
+                match select(event_rx.next(), vdom_fut).await {
+                    Either::Left((l, _)) => l,
+                    Either::Right((_, _)) => None,
+                }
+            };
+
+            if let Some(new_event) = new_event {
+                vdom.handle_message(dioxus_core::SchedulerMsg::Event(new_event));
+            } else {
+                let mutations = vdom.work_with_deadline(|| false);
+                for mutation in mutations {
+                    let edits = serde_json::to_string(&mutation.edits).unwrap();
+                    edits_tx.send(edits).unwrap();
+                }
+            }
+        }
+    });
+
+    loop {
+        use futures_util::future::{select, Either};
+
+        match select(user_ws_rx.next(), edits_rx.next()).await {
+            Either::Left((l, _)) => {
+                if let Some(Ok(msg)) = l {
+                    if let Ok(Some(msg)) = msg.to_str().map(events::parse_ipc_message) {
+                        if msg.method == "user_event" {
+                            let user_event = events::trigger_from_serialized(msg.params);
+                            event_tx.send(user_event).unwrap();
+                        }
+                    } else {
+                        break;
+                    }
+                } else {
+                    break;
+                }
+            }
+            Either::Right((edits, _)) => {
+                if let Some(edits) = edits {
+                    // send the edits to the client
+                    if user_ws_tx.send(Message::text(edits)).await.is_err() {
+                        break;
+                    }
+                } else {
+                    break;
+                }
+            }
+        }
+    }
+
+    vdom_fut.abort();
 }