feat: replace futures channel with flume

Cargo.toml

@@ -16,15 +16,15 @@ compio-log = "0.1.0"
 
 cfg-if = "1"
 euclid = "0.22"
+flume = "0.11"
 futures-util = { version = "0.3", features = ["alloc"] }
-futures-channel = "0.3"
 image = { version = "0.25", default-features = false }
 once_cell = "1"
+paste = "1.0"
 rgb = "0.8"
 slab = "0.4"
 scoped-tls = "1.0"
 taffy = "0.8"
-paste = "1.0"
 
 [target.'cfg(windows)'.dependencies]
 widestring = "1"
@@ -136,14 +136,17 @@ objc2-quartz-core = { version = "0.3", features = [
     "CAShapeLayer",
     "CATextLayer",
 ] }
+local-sync = "0.1"
 
-[target.'cfg(not(any(windows, target_os = "macos")))'.dependencies]
+[target.'cfg(target_os = "linux")'.dependencies]
 libc = "0.2"
 
+[target.'cfg(not(any(windows, target_os = "macos")))'.dependencies]
 gtk4 = { version = "0.9", optional = true, features = ["v4_14"] }
 pangocairo = { version = "0.20", optional = true }
 
 cxx = { version = "1.0", optional = true, features = ["c++17"] }
+local-sync = { version = "0.1", optional = true }
 
 [dev-dependencies]
 compio = { version = "0.15.0", features = ["time", "rustls", "ring"] }
@@ -170,7 +173,7 @@ qt-build-utils = { version = "0.7", optional = true }
 [features]
 default = ["qt", "windows-dark-mode"]
 gtk = ["dep:gtk4", "dep:pangocairo"]
-qt = ["dep:cxx", "dep:cxx-build", "dep:qt-build-utils"]
+qt = ["dep:cxx", "dep:cxx-build", "dep:qt-build-utils", "dep:local-sync"]
 enable_log = ["compio/enable_log", "compio-log/enable_log"]
 
 nightly = ["compio/nightly", "cyper/nightly"]

examples/basic.rs

@@ -58,9 +58,7 @@ impl Component for MainModel {
             let mut interval = interval(Duration::from_secs(1));
             loop {
                 interval.tick().await;
-                if !sender.post(MainMessage::Tick) {
-                    break;
-                }
+                sender.post(MainMessage::Tick);
             }
         })
         .detach();

src/elm/mod.rs

@@ -1,7 +1,6 @@
 use std::{future::Future, hint::unreachable_unchecked};
 
-use futures_channel::mpsc;
-use futures_util::{FutureExt, StreamExt};
+use futures_util::FutureExt;
 
 use crate::{Point, Size, runtime::Runtime};
 
@@ -36,7 +35,7 @@ enum ComponentMessage<T: Component> {
 
 /// Sender of input messages and output events.
 #[derive(Debug)]
-pub struct ComponentSender<T: Component>(mpsc::UnboundedSender<ComponentMessage<T>>);
+pub struct ComponentSender<T: Component>(flume::Sender<ComponentMessage<T>>);
 
 impl<T: Component> Clone for ComponentSender<T> {
     fn clone(&self) -> Self {
@@ -45,36 +44,39 @@ impl<T: Component> Clone for ComponentSender<T> {
 }
 
 #[derive(Debug)]
-struct ComponentReceiver<T: Component>(mpsc::UnboundedReceiver<ComponentMessage<T>>);
+struct ComponentReceiver<T: Component>(flume::Receiver<ComponentMessage<T>>);
 
 fn component_channel<T: Component>() -> (ComponentSender<T>, ComponentReceiver<T>) {
-    let (tx, rx) = mpsc::unbounded();
+    let (tx, rx) = flume::unbounded();
     (ComponentSender(tx), ComponentReceiver(rx))
 }
 
 impl<T: Component> ComponentSender<T> {
     /// Post the message to the queue.
-    pub fn post(&self, message: T::Message) -> bool {
+    pub fn post(&self, message: T::Message) {
         self.0
-            .unbounded_send(ComponentMessage::Message(message))
-            .is_ok()
+            .send(ComponentMessage::Message(message))
+            .expect("component receiver disconnected")
     }
 
     /// Post the event to the queue.
-    pub fn output(&self, event: T::Event) -> bool {
+    pub fn output(&self, event: T::Event) {
         self.0
-            .unbounded_send(ComponentMessage::Event(event))
-            .is_ok()
+            .send(ComponentMessage::Event(event))
+            .expect("component receiver disconnected")
     }
 }
 
 impl<T: Component> ComponentReceiver<T> {
     async fn recv(&mut self) -> ComponentMessage<T> {
-        self.0.next().await.unwrap()
+        self.0
+            .recv_async()
+            .await
+            .expect("component sender disconnected")
     }
 
-    fn try_recv(&mut self) -> Option<ComponentMessage<T>> {
-        self.0.try_next().ok().flatten()
+    fn drain(&mut self) -> impl Iterator<Item = ComponentMessage<T>> + '_ {
+        self.0.drain()
     }
 }
 
@@ -126,7 +128,7 @@ impl App {
                             ComponentMessage::Message(msg) => model.update(msg, &sender).await,
                             ComponentMessage::Event(e) => break 'outer e,
                         };
-                        while let Some(msg) = recv.try_recv() {
+                        for msg in recv.drain() {
                             need_render |= match msg {
                                 ComponentMessage::Message(msg) => model.update(msg, &sender).await,
                                 ComponentMessage::Event(e) => break 'outer e,

src/ui/mac/filebox.rs

@@ -183,7 +183,7 @@ async unsafe fn filebox(
     }
 
     let res = if let Some(parent) = &parent {
-        let (tx, rx) = futures_channel::oneshot::channel();
+        let (tx, rx) = local_sync::oneshot::channel();
         let tx = Rc::new(RefCell::new(Some(tx)));
         let block = StackBlock::new(move |res| {
             tx.borrow_mut()

src/ui/mac/msgbox.rs

@@ -81,7 +81,7 @@ async fn msgbox_custom(
         }
 
         if let Some(parent) = &parent {
-            let (tx, rx) = futures_channel::oneshot::channel();
+            let (tx, rx) = local_sync::oneshot::channel();
             let tx = Rc::new(RefCell::new(Some(tx)));
             let block = StackBlock::new(move |res| {
                 tx.borrow_mut()

src/ui/qt/filebox.rs

@@ -1,7 +1,7 @@
 use std::{mem::ManuallyDrop, path::PathBuf, ptr::null_mut};
 
 use cxx::{ExternType, type_id};
-use futures_channel::oneshot;
+use local_sync::oneshot;
 
 use crate::{AsRawWindow, AsWindow};
 

src/ui/qt/msgbox.rs

@@ -1,7 +1,7 @@
 use std::{collections::HashMap, mem::ManuallyDrop, ptr::null_mut};
 
 use cxx::{ExternType, type_id};
-use futures_channel::oneshot;
+use local_sync::oneshot;
 
 use crate::{AsRawWindow, AsWindow, MessageBoxButton, MessageBoxResponse, MessageBoxStyle};