From 959e292bd51bbf6df1b667ee0921a93cba117cbb Mon Sep 17 00:00:00 2001
From: Jared Moulton <jaredmoulton3@gmail.com>
Date: Fri, 31 May 2024 02:28:11 -0700
Subject: [PATCH] Add tokio channel behind flag (#483)

---
 Cargo.toml       |  3 +++
 src/ext_event.rs | 40 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 43 insertions(+)

diff --git a/Cargo.toml b/Cargo.toml
index a9e20d4b..d4022d9e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -43,6 +43,7 @@ taffy = { version = "0.4", features = ["grid"] }
 rfd = { version = "0.14.0", default-features = false, features = [
     "xdg-portal",
 ], optional = true }
+tokio = { version = "1", features = ["sync", "rt"], optional = true }
 raw-window-handle = { workspace = true }
 unicode-segmentation = "1.10.0"
 peniko = { workspace = true }
@@ -93,6 +94,8 @@ image-tga = ["image/tga"]
 image-tiff = ["image/tiff"]
 image-webp = ["image/webp"]
 
+tokio = ["dep:tokio"]
+
 # rfd (file dialog) async runtime
 rfd-async-std = ["dep:rfd", "rfd/async-std"]
 rfd-tokio = ["dep:rfd", "rfd/tokio"]
diff --git a/src/ext_event.rs b/src/ext_event.rs
index aa43d9c7..12f1174f 100644
--- a/src/ext_event.rs
+++ b/src/ext_event.rs
@@ -148,3 +148,43 @@ pub fn create_signal_from_channel<T: Send + 'static>(
 
     read
 }
+
+#[cfg(feature = "tokio")]
+pub fn create_signal_from_tokio_channel<T: Send + 'static>(
+    mut rx: tokio::sync::mpsc::UnboundedReceiver<T>,
+) -> ReadSignal<Option<T>> {
+    let cx = Scope::new();
+    let trigger = cx.create_trigger();
+
+    let channel_closed = cx.create_rw_signal(false);
+    let (read, write) = cx.create_signal(None);
+    let data = std::sync::Arc::new(std::sync::Mutex::new(VecDeque::new()));
+
+    {
+        let data = data.clone();
+        cx.create_effect(move |_| {
+            trigger.track();
+            while let Some(value) = data.lock().unwrap().pop_front() {
+                write.set(value);
+            }
+
+            if channel_closed.get() {
+                cx.dispose();
+            }
+        });
+    }
+
+    let send = create_ext_action(cx, move |_| {
+        channel_closed.set(true);
+    });
+
+    tokio::spawn(async move {
+        while let Some(event) = rx.recv().await {
+            data.lock().unwrap().push_back(Some(event));
+            crate::ext_event::register_ext_trigger(trigger);
+        }
+        send(());
+    });
+
+    read
+}