1 files changed, 74 insertions, 1 deletions

diff --git a/winit/src/application.rs b/winit/src/application.rs
index 85d06d9b..26ebdb05 100644
--- a/winit/src/application.rs
+++ b/winit/src/application.rs
@@ -3,8 +3,9 @@ use crate::{
     input::{keyboard, mouse},
     renderer::{Target, Windowed},
     Cache, Command, Container, Debug, Element, Event, Length, MouseCursor,
-    Settings, UserInterface,
+    Settings, Subscription, UserInterface,
 };
+use std::collections::HashMap;
 
 /// An interactive, native cross-platform application.
 ///
@@ -57,6 +58,9 @@ pub trait Application: Sized {
     /// [`Command`]: struct.Command.html
     fn update(&mut self, message: Self::Message) -> Command<Self::Message>;
 
+    /// TODO
+    fn subscriptions(&self) -> Subscription<Self::Message>;
+
     /// Returns the widgets to display in the [`Application`].
     ///
     /// These widgets can produce __messages__ based on user interaction.
@@ -89,11 +93,15 @@ pub trait Application: Sized {
         let proxy = event_loop.create_proxy();
         let mut thread_pool =
             futures::executor::ThreadPool::new().expect("Create thread pool");
+        let mut alive_subscriptions = Subscriptions::new();
         let mut external_messages = Vec::new();
 
         let (mut application, init_command) = Self::new();
         spawn(init_command, &mut thread_pool, &proxy);
 
+        let subscriptions = application.subscriptions();
+        alive_subscriptions.update(subscriptions, &mut thread_pool, &proxy);
+
         let mut title = application.title();
 
         let window = {
@@ -204,6 +212,13 @@ pub trait Application: Sized {
                         debug.update_finished();
                     }
 
+                    let subscriptions = application.subscriptions();
+                    alive_subscriptions.update(
+                        subscriptions,
+                        &mut thread_pool,
+                        &proxy,
+                    );
+
                     // Update window title
                     let new_title = application.title();
 
@@ -406,6 +421,64 @@ fn spawn<Message: Send>(
     }
 }
 
+pub struct Subscriptions {
+    alive: HashMap<u64, futures::channel::oneshot::Sender<()>>,
+}
+
+impl Subscriptions {
+    fn new() -> Self {
+        Self {
+            alive: HashMap::new(),
+        }
+    }
+
+    fn update<Message: Send>(
+        &mut self,
+        subscriptions: Subscription<Message>,
+        thread_pool: &mut futures::executor::ThreadPool,
+        proxy: &winit::event_loop::EventLoopProxy<Message>,
+    ) {
+        use futures::{future::FutureExt, stream::StreamExt};
+
+        let handles = subscriptions.handles();
+        let mut alive = std::collections::HashSet::new();
+
+        for handle in handles {
+            let id = handle.id();
+            let _ = alive.insert(id);
+
+            if !self.alive.contains_key(&id) {
+                let (cancel, cancelled) = futures::channel::oneshot::channel();
+
+                let stream = handle.stream();
+
+                let proxy =
+                    std::sync::Arc::new(std::sync::Mutex::new(proxy.clone()));
+
+                let future = futures::future::select(
+                    cancelled,
+                    stream.for_each(move |message| {
+                        proxy
+                            .lock()
+                            .expect("Acquire event loop proxy lock")
+                            .send_event(message)
+                            .expect("Send subscription result to event loop");
+
+                        futures::future::ready(())
+                    }),
+                )
+                .map(|_| ());
+
+                thread_pool.spawn_ok(future);
+
+                let _ = self.alive.insert(id, cancel);
+            }
+        }
+
+        self.alive.retain(|id, _| alive.contains(&id));
+    }
+}
+
 // As defined in: http://www.unicode.org/faq/private_use.html
 // TODO: Remove once https://github.com/rust-windowing/winit/pull/1254 lands
 fn is_private_use_character(c: char) -> bool {