On Web, implement Send for EventLoopProxy

CHANGELOG.md

@@ -13,6 +13,7 @@ And please only add new entries to the top of this list, right below the `# Unre
 # 0.28.3
 
 - Fix macOS memory leaks.
+- On Web, `EventLoopProxy` now implements `Send`.
 
 # 0.28.2
 

Cargo.toml

@@ -148,8 +148,9 @@ features = [
     'WheelEvent'
 ]
 
-[target.'cfg(target_family = "wasm")'.dependencies.wasm-bindgen]
-version = "0.2.45"
+[target.'cfg(target_family = "wasm")'.dependencies]
+wasm-bindgen = "0.2.45"
+wasm-bindgen-futures = "0.4.31"
 
 [target.'cfg(target_family = "wasm")'.dev-dependencies]
 console_log = "0.2"

src/platform_impl/web/event_loop/proxy.rs

@@ -1,26 +1,119 @@
-use super::runner;
-use crate::event::Event;
+use std::future;
+use std::rc::Rc;
+use std::sync::mpsc::{self, Receiver, RecvError, SendError, Sender, TryRecvError};
+use std::sync::{Arc, Mutex};
+use std::task::{Poll, Waker};
+
 use crate::event_loop::EventLoopClosed;
 
 pub struct EventLoopProxy<T: 'static> {
-    runner: runner::Shared<T>,
+    sender: AsyncSender<T>,
 }
 
 impl<T: 'static> EventLoopProxy<T> {
-    pub fn new(runner: runner::Shared<T>) -> Self {
-        Self { runner }
+    pub fn new(sender: AsyncSender<T>) -> Self {
+        Self { sender }
     }
 
     pub fn send_event(&self, event: T) -> Result<(), EventLoopClosed<T>> {
-        self.runner.send_event(Event::UserEvent(event));
-        Ok(())
+        match self.sender.send(event) {
+            Ok(()) => Ok(()),
+            Err(SendError(val)) => Err(EventLoopClosed(val)),
+        }
     }
 }
 
 impl<T: 'static> Clone for EventLoopProxy<T> {
     fn clone(&self) -> Self {
         Self {
-            runner: self.runner.clone(),
+            sender: self.sender.clone(),
+        }
+    }
+}
+
+pub fn channel<T: 'static>() -> (AsyncSender<T>, AsyncReceiver<T>) {
+    let (sender, receiver) = mpsc::channel();
+    let waker = Arc::new(Mutex::new(None));
+
+    let sender = AsyncSender {
+        sender: Some(sender),
+        waker: Arc::clone(&waker),
+    };
+    let receiver = AsyncReceiver {
+        receiver: Rc::new(receiver),
+        waker,
+    };
+
+    (sender, receiver)
+}
+
+pub struct AsyncSender<T: 'static> {
+    sender: Option<Sender<T>>,
+    waker: Arc<Mutex<Option<Waker>>>,
+}
+
+impl<T: 'static> AsyncSender<T> {
+    pub fn send(&self, event: T) -> Result<(), SendError<T>> {
+        self.sender.as_ref().unwrap().send(event)?;
+
+        if let Some(waker) = self.waker.lock().unwrap().take() {
+            waker.wake();
+        }
+
+        Ok(())
+    }
+}
+
+impl<T: 'static> Clone for AsyncSender<T> {
+    fn clone(&self) -> Self {
+        Self {
+            sender: self.sender.clone(),
+            waker: self.waker.clone(),
         }
     }
 }
+
+impl<T> Drop for AsyncSender<T> {
+    fn drop(&mut self) {
+        // The corresponding `Receiver` is used to spawn a future that waits
+        // for messages in a loop. The future itself needs to be cleaned up
+        // somehow, which is signalled by dropping the last `Sender`. But it
+        // will do nothing if not woken up.
+
+        // We have to drop the potentially last `Sender` **before** checking if
+        // this is the last `Sender`. `Arc::strong_count` doesn't prevent
+        // races.
+        self.sender.take().unwrap();
+
+        // This one + the one held by the future.
+        if Arc::strong_count(&self.waker) == 2 {
+            if let Some(waker) = self.waker.lock().unwrap().take() {
+                waker.wake();
+            }
+        }
+    }
+}
+
+pub struct AsyncReceiver<T: 'static> {
+    receiver: Rc<Receiver<T>>,
+    waker: Arc<Mutex<Option<Waker>>>,
+}
+
+impl<T: 'static> AsyncReceiver<T> {
+    pub async fn next(&mut self) -> Result<T, RecvError> {
+        future::poll_fn(|cx| match self.receiver.try_recv() {
+            Ok(event) => Poll::Ready(Ok(event)),
+            Err(TryRecvError::Empty) => {
+                *self.waker.lock().unwrap() = Some(cx.waker().clone());
+
+                match self.receiver.try_recv() {
+                    Ok(event) => Poll::Ready(Ok(event)),
+                    Err(TryRecvError::Empty) => Poll::Pending,
+                    Err(TryRecvError::Disconnected) => Poll::Ready(Err(RecvError)),
+                }
+            }
+            Err(TryRecvError::Disconnected) => Poll::Ready(Err(RecvError)),
+        })
+        .await
+    }
+}

src/platform_impl/web/event_loop/runner.rs

@@ -1,3 +1,5 @@
+use super::proxy::{self, AsyncSender};
+use super::EventLoopProxy;
 use super::{super::ScaleChangeArgs, backend, state::State};
 use crate::event::{Event, StartCause};
 use crate::event_loop::ControlFlow;
@@ -32,6 +34,7 @@ pub struct Execution<T: 'static> {
     destroy_pending: RefCell<VecDeque<WindowId>>,
     scale_change_detector: RefCell<Option<backend::ScaleChangeDetector>>,
     unload_event_handle: RefCell<Option<backend::UnloadEventHandle>>,
+    proxy_sender: AsyncSender<T>,
 }
 
 enum RunnerEnum<T: 'static> {
@@ -99,7 +102,9 @@ impl<T: 'static> Runner<T> {
 
 impl<T: 'static> Shared<T> {
     pub fn new() -> Self {
-        Shared(Rc::new(Execution {
+        let (proxy_sender, mut proxy_receiver) = proxy::channel();
+
+        let this = Shared(Rc::new(Execution {
             runner: RefCell::new(RunnerEnum::Pending),
             events: RefCell::new(VecDeque::new()),
             id: RefCell::new(0),
@@ -108,7 +113,22 @@ impl<T: 'static> Shared<T> {
             destroy_pending: RefCell::new(VecDeque::new()),
             scale_change_detector: RefCell::new(None),
             unload_event_handle: RefCell::new(None),
-        }))
+            proxy_sender,
+        }));
+
+        wasm_bindgen_futures::spawn_local({
+            let runner = this.clone();
+            async move {
+                while let Ok(value) = proxy_receiver.next().await {
+                    runner.send_event(Event::UserEvent(value))
+                }
+
+                // An error was returned because the channel was closed, which
+                // happens when the event loop gets closed, so we can stop now.
+            }
+        });
+
+        this
     }
 
     pub fn add_canvas(&self, id: WindowId, canvas: &Rc<RefCell<backend::Canvas>>) {
@@ -155,6 +175,10 @@ impl<T: 'static> Shared<T> {
         *id
     }
 
+    pub fn create_proxy(&self) -> EventLoopProxy<T> {
+        EventLoopProxy::new(self.0.proxy_sender.clone())
+    }
+
     pub fn request_redraw(&self, id: WindowId) {
         self.0.redraw_pending.borrow_mut().insert(id);
     }

src/platform_impl/web/event_loop/window_target.rs

@@ -36,7 +36,7 @@ impl<T> EventLoopWindowTarget<T> {
     }
 
     pub fn proxy(&self) -> EventLoopProxy<T> {
-        EventLoopProxy::new(self.runner.clone())
+        self.runner.create_proxy()
     }
 
     pub fn run(&self, event_handler: Box<runner::EventHandler<T>>) {

tests/send_objects.rs

@@ -1,7 +1,6 @@
 #[allow(dead_code)]
 fn needs_send<T: Send>() {}
 
-#[cfg(not(wasm_platform))]
 #[test]
 fn event_loop_proxy_send() {
     #[allow(dead_code)]