Convert use_eval to use send/recv system (#1080)

* progress: reworked don't run this, it'll kill your web browser * feat: use_eval but with comms * revision: async recv & recv_sync * revision: use_eval * revision: standard eval interface * revision: use serde_json::Value instead of JsValue * revision: docs * revision: error message * create: desktop eval (wip) * fix: desktop eval * revision: wrap use_eval in Rc<RefCell<_>> * fix: fmt, clippy * fix: desktop tests * revision: change to channel system - fixes clippy errors - fixes playwright tests * fix: tests * fix: eval example * fix: fmt * fix: tests, desktop stuff * fix: tests * feat: drop handler * fix: tests * fix: rustfmt * revision: web promise/callback system * fix: recv error * revision: IntoFuture, functionation * fix: ci * revision: playwright web * remove: unescessary code * remove dioxus-html from public examples * prototype-patch * fix web eval * fix: rustfmt * fix: CI * make use_eval more efficient * implement eval for liveview as well * fix playwright tests * fix clippy * more clippy fixes * fix clippy * fix stack overflow * fix desktop mock * fix clippy --------- Co-authored-by: Evan Almloff <evanalmloff@gmail.com>
2023-07-21 17:36:25 -05:00
1 parent b36a7a3993
commit 6210c6fefe
@@ -1,23 +1,17 @@
 use dioxus_core::ElementId;
 use dioxus_html::{geometry::euclid::Rect, MountedResult, RenderedElementBacking};
-use tokio::sync::mpsc::UnboundedSender;

 use crate::query::QueryEngine;

 /// A mounted element passed to onmounted events
 pub struct LiveviewElement {
    id: ElementId,
-    query_tx: UnboundedSender<String>,
    query: QueryEngine,
 }

 impl LiveviewElement {
-    pub(crate) fn new(id: ElementId, tx: UnboundedSender<String>, query: QueryEngine) -> Self {
-        Self {
-            id,
-            query_tx: tx,
-            query,
-        }
+    pub(crate) fn new(id: ElementId, query: QueryEngine) -> Self {
+        Self { id, query }
    }
 }

@@ -39,7 +33,7 @@ impl RenderedElementBacking for LiveviewElement {

        let fut = self
            .query
-            .new_query::<Option<Rect<f64, f64>>>(&script, &self.query_tx)
+            .new_query::<Option<Rect<f64, f64>>>(&script)
            .resolve();
        Box::pin(async move {
            match fut.await {
@@ -64,10 +58,7 @@ impl RenderedElementBacking for LiveviewElement {
            serde_json::to_string(&behavior).expect("Failed to serialize ScrollBehavior")
        );

-        let fut = self
-            .query
-            .new_query::<bool>(&script, &self.query_tx)
-            .resolve();
+        let fut = self.query.new_query::<bool>(&script).resolve();
        Box::pin(async move {
            match fut.await {
                Ok(true) => Ok(()),
@@ -90,10 +81,7 @@ impl RenderedElementBacking for LiveviewElement {
            self.id.0, focus
        );

-        let fut = self
-            .query
-            .new_query::<bool>(&script, &self.query_tx)
-            .resolve();
+        let fut = self.query.new_query::<bool>(&script).resolve();

        Box::pin(async move {
            match fut.await {
@@ -0,0 +1,75 @@
+#![allow(clippy::await_holding_refcell_ref)]
+
+use async_trait::async_trait;
+use dioxus_core::ScopeState;
+use dioxus_html::prelude::{EvalError, EvalProvider, Evaluator};
+use std::{cell::RefCell, rc::Rc};
+
+use crate::query::{Query, QueryEngine};
+
+/// Provides the DesktopEvalProvider through [`cx.provide_context`].
+pub fn init_eval(cx: &ScopeState) {
+    let query = cx.consume_context::<QueryEngine>().unwrap();
+    let provider: Rc<dyn EvalProvider> = Rc::new(DesktopEvalProvider { query });
+    cx.provide_context(provider);
+}
+
+/// Reprents the desktop-target's provider of evaluators.
+pub struct DesktopEvalProvider {
+    query: QueryEngine,
+}
+
+impl EvalProvider for DesktopEvalProvider {
+    fn new_evaluator(&self, js: String) -> Result<Rc<dyn Evaluator>, EvalError> {
+        Ok(Rc::new(DesktopEvaluator::new(self.query.clone(), js)))
+    }
+}
+
+/// Reprents a desktop-target's JavaScript evaluator.
+pub(crate) struct DesktopEvaluator {
+    query: Rc<RefCell<Query<serde_json::Value>>>,
+}
+
+impl DesktopEvaluator {
+    /// Creates a new evaluator for desktop-based targets.
+    pub fn new(query: QueryEngine, js: String) -> Self {
+        let query = query.new_query(&js);
+
+        Self {
+            query: Rc::new(RefCell::new(query)),
+        }
+    }
+}
+
+#[async_trait(?Send)]
+impl Evaluator for DesktopEvaluator {
+    /// # Panics
+    /// This will panic if the query is currently being awaited.
+    async fn join(&self) -> Result<serde_json::Value, EvalError> {
+        self.query
+            .borrow_mut()
+            .result()
+            .await
+            .map_err(|e| EvalError::Communication(e.to_string()))
+    }
+
+    /// Sends a message to the evaluated JavaScript.
+    fn send(&self, data: serde_json::Value) -> Result<(), EvalError> {
+        if let Err(e) = self.query.borrow_mut().send(data) {
+            return Err(EvalError::Communication(e.to_string()));
+        }
+        Ok(())
+    }
+
+    /// Gets an UnboundedReceiver to receive messages from the evaluated JavaScript.
+    ///
+    /// # Panics
+    /// This will panic if the query is currently being awaited.
+    async fn recv(&self) -> Result<serde_json::Value, EvalError> {
+        self.query
+            .borrow_mut()
+            .recv()
+            .await
+            .map_err(|e| EvalError::Communication(e.to_string()))
+    }
+}
@@ -23,6 +23,7 @@ pub mod pool;
 mod query;
 use futures_util::{SinkExt, StreamExt};
 pub use pool::*;
+mod eval;

 pub trait WebsocketTx: SinkExt<String, Error = LiveViewError> {}
 impl<T> WebsocketTx for T where T: SinkExt<String, Error = LiveViewError> {}
@@ -1,5 +1,6 @@
 use crate::{
    element::LiveviewElement,
+    eval::init_eval,
    query::{QueryEngine, QueryResult},
    LiveViewError,
 };
@@ -119,6 +120,12 @@ pub async fn run(mut vdom: VirtualDom, ws: impl LiveViewSocket) -> Result<(), Li
        rx
    };

+    // Create the a proxy for query engine
+    let (query_tx, mut query_rx) = tokio::sync::mpsc::unbounded_channel();
+    let query_engine = QueryEngine::new(query_tx);
+    vdom.base_scope().provide_context(query_engine.clone());
+    init_eval(vdom.base_scope());
+
    // todo: use an efficient binary packed format for this
    let edits = serde_json::to_string(&ClientUpdate::Edits(vdom.rebuild())).unwrap();

@@ -128,10 +135,6 @@ pub async fn run(mut vdom: VirtualDom, ws: impl LiveViewSocket) -> Result<(), Li
    // send the initial render to the client
    ws.send(edits.into_bytes()).await?;

-    // Create the a proxy for query engine
-    let (query_tx, mut query_rx) = tokio::sync::mpsc::unbounded_channel();
-    let query_engine = QueryEngine::default();
-
    // desktop uses this wrapper struct thing around the actual event itself
    // this is sorta driven by tao/wry
    #[derive(serde::Deserialize, Debug)]
@@ -165,7 +168,7 @@ pub async fn run(mut vdom: VirtualDom, ws: impl LiveViewSocket) -> Result<(), Li
                                IpcMessage::Event(evt) => {
                                    // Intercept the mounted event and insert a custom element type
                                    if let EventData::Mounted = &evt.data {
-                                        let element = LiveviewElement::new(evt.element, query_tx.clone(), query_engine.clone());
+                                        let element = LiveviewElement::new(evt.element, query_engine.clone());
                                        vdom.handle_event(
                                            &evt.name,
                                            Rc::new(MountedData::new(element)),
@@ -4,110 +4,261 @@ use serde::{de::DeserializeOwned, Deserialize};
 use serde_json::Value;
 use slab::Slab;
 use thiserror::Error;
-use tokio::sync::{broadcast::error::RecvError, mpsc::UnboundedSender};
+use tokio::sync::broadcast::error::RecvError;
+
+const DIOXUS_CODE: &str = r#"
+let dioxus = {
+    recv: function () {
+        return new Promise((resolve, _reject) => {
+            // Ever 50 ms check for new data
+            let timeout = setTimeout(() => {
+                let __msg = null;
+                while (true) {
+                    let __data = _message_queue.shift();
+                    if (__data) {
+                        __msg = __data;
+                        break;
+                    }
+                }
+                clearTimeout(timeout);
+                resolve(__msg);
+            }, 50);
+        });
+    },
+
+    send: function (value) {
+        window.ipc.postMessage(
+            JSON.stringify({
+                "method":"query",
+                "params": {
+                    "id": _request_id,
+                    "data": value,
+                    "returned_value": false
+                }
+            })
+        );
+    }
+}"#;

 /// Tracks what query ids are currently active
-#[derive(Default, Clone)]
-struct SharedSlab {
-    slab: Rc<RefCell<Slab<()>>>,
+
+struct SharedSlab<T = ()> {
+    slab: Rc<RefCell<Slab<T>>>,
 }

-/// Handles sending and receiving arbitrary queries from the webview. Queries can be resolved non-sequentially, so we use ids to track them.
-#[derive(Clone)]
-pub(crate) struct QueryEngine {
-    sender: Rc<tokio::sync::broadcast::Sender<QueryResult>>,
-    active_requests: SharedSlab,
-}
-
-impl Default for QueryEngine {
-    fn default() -> Self {
-        let (sender, _) = tokio::sync::broadcast::channel(8);
+impl<T> Clone for SharedSlab<T> {
+    fn clone(&self) -> Self {
        Self {
-            sender: Rc::new(sender),
-            active_requests: SharedSlab::default(),
+            slab: self.slab.clone(),
        }
    }
 }

+impl<T> Default for SharedSlab<T> {
+    fn default() -> Self {
+        SharedSlab {
+            slab: Rc::new(RefCell::new(Slab::new())),
+        }
+    }
+}
+
+struct QueryEntry {
+    channel_sender: tokio::sync::mpsc::UnboundedSender<Value>,
+    return_sender: Option<tokio::sync::oneshot::Sender<Value>>,
+}
+
+const QUEUE_NAME: &str = "__msg_queues";
+
+/// Handles sending and receiving arbitrary queries from the webview. Queries can be resolved non-sequentially, so we use ids to track them.
+#[derive(Clone)]
+pub(crate) struct QueryEngine {
+    active_requests: SharedSlab<QueryEntry>,
+    query_tx: tokio::sync::mpsc::UnboundedSender<String>,
+}
+
 impl QueryEngine {
+    pub(crate) fn new(query_tx: tokio::sync::mpsc::UnboundedSender<String>) -> Self {
+        Self {
+            active_requests: Default::default(),
+            query_tx,
+        }
+    }
+
    /// Creates a new query and returns a handle to it. The query will be resolved when the webview returns a result with the same id.
-    pub fn new_query<V: DeserializeOwned>(
-        &self,
-        script: &str,
-        tx: &UnboundedSender<String>,
-    ) -> Query<V> {
-        let request_id = self.active_requests.slab.borrow_mut().insert(());
+    pub fn new_query<V: DeserializeOwned>(&self, script: &str) -> Query<V> {
+        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+        let (return_tx, return_rx) = tokio::sync::oneshot::channel();
+        let request_id = self.active_requests.slab.borrow_mut().insert(QueryEntry {
+            channel_sender: tx,
+            return_sender: Some(return_tx),
+        });

        // start the query
        // We embed the return of the eval in a function so we can send it back to the main thread
-        if let Err(err) = tx.send(format!(
-            r#"window.ipc.postMessage(
-                JSON.stringify({{
-                    "method":"query",
-                    "params": {{
-                        "id": {request_id},
-                        "data": (function(){{{script}}})()
+        if let Err(err) = self.query_tx.send(format!(
+            r#"(function(){{
+                (async (resolve, _reject) => {{
+                    {DIOXUS_CODE}
+                    if (!window.{QUEUE_NAME}) {{
+                        window.{QUEUE_NAME} = [];
                    }}
+    
+                    let _request_id = {request_id};
+    
+                    if (!window.{QUEUE_NAME}[{request_id}]) {{
+                        window.{QUEUE_NAME}[{request_id}] = [];
+                    }}
+                    let _message_queue = window.{QUEUE_NAME}[{request_id}];
+    
+                    {script}
+                }})().then((result)=>{{
+                    let returned_value = {{
+                        "method":"query",
+                        "params": {{
+                            "id": {request_id},
+                            "data": result,
+                            "returned_value": true
+                        }}
+                    }};
+                    window.ipc.postMessage(
+                        JSON.stringify(returned_value)
+                    );
                }})
-            );"#
+            }})();"#
        )) {
            log::warn!("Query error: {err}");
        }

        Query {
-            slab: self.active_requests.clone(),
+            query_engine: self.clone(),
            id: request_id,
-            reciever: self.sender.subscribe(),
+            receiver: rx,
+            return_receiver: Some(return_rx),
            phantom: std::marker::PhantomData,
        }
    }

-    /// Send a query result
+    /// Send a query channel message to the correct query
    pub fn send(&self, data: QueryResult) {
-        let _ = self.sender.send(data);
+        let QueryResult {
+            id,
+            data,
+            returned_value,
+        } = data;
+        let mut slab = self.active_requests.slab.borrow_mut();
+        if let Some(entry) = slab.get_mut(id) {
+            if returned_value {
+                if let Some(sender) = entry.return_sender.take() {
+                    let _ = sender.send(data);
+                }
+            } else {
+                let _ = entry.channel_sender.send(data);
+            }
+        }
    }
 }

 pub(crate) struct Query<V: DeserializeOwned> {
-    slab: SharedSlab,
+    query_engine: QueryEngine,
+    pub receiver: tokio::sync::mpsc::UnboundedReceiver<Value>,
+    pub return_receiver: Option<tokio::sync::oneshot::Receiver<Value>>,
    id: usize,
-    reciever: tokio::sync::broadcast::Receiver<QueryResult>,
    phantom: std::marker::PhantomData<V>,
 }

 impl<V: DeserializeOwned> Query<V> {
    /// Resolve the query
    pub async fn resolve(mut self) -> Result<V, QueryError> {
-        let result = loop {
-            match self.reciever.recv().await {
-                Ok(result) => {
-                    if result.id == self.id {
-                        break V::deserialize(result.data).map_err(QueryError::DeserializeError);
-                    }
-                }
-                Err(err) => {
-                    break Err(QueryError::RecvError(err));
-                }
-            }
-        };
+        match self.receiver.recv().await {
+            Some(result) => V::deserialize(result).map_err(QueryError::Deserialize),
+            None => Err(QueryError::Recv(RecvError::Closed)),
+        }
+    }

-        // Remove the query from the slab
-        self.slab.slab.borrow_mut().remove(self.id);
+    /// Send a message to the query
+    pub fn send<S: ToString>(&self, message: S) -> Result<(), QueryError> {
+        let queue_id = self.id;

-        result
+        let data = message.to_string();
+        let script = format!(
+            r#"
+            if (!window.{QUEUE_NAME}) {{
+                window.{QUEUE_NAME} = [];
+            }}
+
+            if (!window.{QUEUE_NAME}[{queue_id}]) {{
+                window.{QUEUE_NAME}[{queue_id}] = [];
+            }}
+            window.{QUEUE_NAME}[{queue_id}].push({data});
+            "#
+        );
+
+        self.query_engine
+            .query_tx
+            .send(script)
+            .map_err(|e| QueryError::Send(e.to_string()))?;
+
+        Ok(())
+    }
+
+    /// Receive a message from the query
+    pub async fn recv(&mut self) -> Result<Value, QueryError> {
+        self.receiver
+            .recv()
+            .await
+            .ok_or(QueryError::Recv(RecvError::Closed))
+    }
+
+    /// Receive the result of the query
+    pub async fn result(&mut self) -> Result<Value, QueryError> {
+        match self.return_receiver.take() {
+            Some(receiver) => receiver
+                .await
+                .map_err(|_| QueryError::Recv(RecvError::Closed)),
+            None => Err(QueryError::Finished),
+        }
+    }
+}
+
+impl<V: DeserializeOwned> Drop for Query<V> {
+    fn drop(&mut self) {
+        self.query_engine
+            .active_requests
+            .slab
+            .borrow_mut()
+            .remove(self.id);
+        let queue_id = self.id;
+
+        _ = self.query_engine.query_tx.send(format!(
+            r#"
+            if (!window.{QUEUE_NAME}) {{
+                window.{QUEUE_NAME} = [];
+            }}
+
+            if (window.{QUEUE_NAME}[{queue_id}]) {{
+                window.{QUEUE_NAME}[{queue_id}] = [];
+            }}
+            "#
+        ));
    }
 }

 #[derive(Error, Debug)]
 pub enum QueryError {
    #[error("Error receiving query result: {0}")]
-    RecvError(RecvError),
+    Recv(RecvError),
+    #[error("Error sending message to query: {0}")]
+    Send(String),
    #[error("Error deserializing query result: {0}")]
-    DeserializeError(serde_json::Error),
+    Deserialize(serde_json::Error),
+    #[error("Query has already been resolved")]
+    Finished,
 }

 #[derive(Clone, Debug, Deserialize)]
 pub(crate) struct QueryResult {
    id: usize,
    data: Value,
+    #[serde(default)]
+    returned_value: bool,
 }