Allow listening to runtime events in subscriptions

7 files changed, 89 insertions, 40 deletions

diff --git a/core/src/subscription.rs b/core/src/subscription.rs
index 796982c7..4c021d75 100644
--- a/core/src/subscription.rs
+++ b/core/src/subscription.rs
@@ -1,51 +1,59 @@
 //! Generate events asynchronously for you application.
 
 /// An event subscription.
-pub struct Subscription<T> {
-    handles: Vec<Box<dyn Handle<Output = T>>>,
+pub struct Subscription<I, O> {
+    connections: Vec<Box<dyn Connection<Input = I, Output = O>>>,
 }
 
-impl<T> Subscription<T> {
+impl<I, O> Subscription<I, O> {
     pub fn none() -> Self {
         Self {
-            handles: Vec::new(),
+            connections: Vec::new(),
         }
     }
 
-    pub fn batch(subscriptions: impl Iterator<Item = Subscription<T>>) -> Self {
+    pub fn batch(
+        subscriptions: impl Iterator<Item = Subscription<I, O>>,
+    ) -> Self {
         Self {
-            handles: subscriptions
-                .flat_map(|subscription| subscription.handles)
+            connections: subscriptions
+                .flat_map(|subscription| subscription.connections)
                 .collect(),
         }
     }
 
-    pub fn handles(self) -> Vec<Box<dyn Handle<Output = T>>> {
-        self.handles
+    pub fn connections(
+        self,
+    ) -> Vec<Box<dyn Connection<Input = I, Output = O>>> {
+        self.connections
     }
 }
 
-impl<T, A> From<A> for Subscription<T>
+impl<I, O, T> From<T> for Subscription<I, O>
 where
-    A: Handle<Output = T> + 'static,
+    T: Connection<Input = I, Output = O> + 'static,
 {
-    fn from(handle: A) -> Self {
+    fn from(handle: T) -> Self {
         Self {
-            handles: vec![Box::new(handle)],
+            connections: vec![Box::new(handle)],
         }
     }
 }
 
-/// The handle of an event subscription.
-pub trait Handle {
+/// The connection of an event subscription.
+pub trait Connection {
+    type Input;
     type Output;
 
     fn id(&self) -> u64;
 
-    fn stream(&self) -> futures::stream::BoxStream<'static, Self::Output>;
+    fn stream(
+        &self,
+        input: Self::Input,
+    ) -> futures::stream::BoxStream<'static, Self::Output>;
 }
 
-impl<T> std::fmt::Debug for Subscription<T> {
+impl<I, O> std::fmt::Debug for Subscription<I, O> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Subscription").finish()
     }
diff --git a/examples/clock.rs b/examples/clock.rs
index 5a404bfa..f06762bd 100644
--- a/examples/clock.rs
+++ b/examples/clock.rs
@@ -4,6 +4,8 @@ use iced::{
 };
 
 pub fn main() {
+    env_logger::init();
+
     Clock::run(Settings::default())
 }
 
@@ -108,30 +110,26 @@ mod time {
         >,
     }
 
-    impl<Message> iced::subscription::Handle for Tick<Message>
+    impl<Message> iced_native::subscription::Connection for Tick<Message>
     where
         Message: 'static,
     {
+        type Input = iced_native::subscription::Input;
         type Output = Message;
 
         fn id(&self) -> u64 {
             0
         }
 
-        fn stream(&self) -> futures::stream::BoxStream<'static, Message> {
+        fn stream(
+            &self,
+            input: iced_native::subscription::Input,
+        ) -> futures::stream::BoxStream<'static, Message> {
             use futures::StreamExt;
 
-            let duration = self.duration.clone();
             let function = self.message.clone();
 
-            let stream =
-                futures::stream::iter(std::iter::repeat(())).map(move |_| {
-                    std::thread::sleep(duration);
-
-                    function(chrono::Local::now())
-                });
-
-            stream.boxed()
+            input.map(move |_| function(chrono::Local::now())).boxed()
         }
     }
 }
diff --git a/native/Cargo.toml b/native/Cargo.toml
index 7007855a..a31b6627 100644
--- a/native/Cargo.toml
+++ b/native/Cargo.toml
@@ -12,3 +12,4 @@ iced_core = { version = "0.1.0", path = "../core", features = ["command", "subsc
 twox-hash = "1.5"
 raw-window-handle = "0.3"
 unicode-segmentation = "1.6"
+futures = "0.3"
diff --git a/native/src/lib.rs b/native/src/lib.rs
index 9afb3bc9..af937a2f 100644
--- a/native/src/lib.rs
+++ b/native/src/lib.rs
@@ -34,7 +34,7 @@
 //! [`Windowed`]: renderer/trait.Windowed.html
 //! [`UserInterface`]: struct.UserInterface.html
 //! [renderer]: renderer/index.html
-#![deny(missing_docs)]
+//#![deny(missing_docs)]
 #![deny(missing_debug_implementations)]
 #![deny(unused_results)]
 #![deny(unsafe_code)]
@@ -42,6 +42,7 @@
 pub mod input;
 pub mod layout;
 pub mod renderer;
+pub mod subscription;
 pub mod widget;
 
 mod element;
@@ -52,8 +53,8 @@ mod size;
 mod user_interface;
 
 pub use iced_core::{
-    subscription, Align, Background, Color, Command, Font, HorizontalAlignment,
-    Length, Point, Rectangle, Subscription, Vector, VerticalAlignment,
+    Align, Background, Color, Command, Font, HorizontalAlignment, Length,
+    Point, Rectangle, Vector, VerticalAlignment,
 };
 
 pub use element::Element;
@@ -63,5 +64,6 @@ pub use layout::Layout;
 pub use mouse_cursor::MouseCursor;
 pub use renderer::Renderer;
 pub use size::Size;
+pub use subscription::Subscription;
 pub use user_interface::{Cache, UserInterface};
 pub use widget::*;
diff --git a/native/src/subscription.rs b/native/src/subscription.rs
new file mode 100644
index 00000000..5fa026a2
--- /dev/null
+++ b/native/src/subscription.rs
@@ -0,0 +1,6 @@
+use crate::Event;
+
+pub type Subscription<T> = iced_core::Subscription<Input, T>;
+pub type Input = futures::channel::mpsc::Receiver<Event>;
+
+pub use iced_core::subscription::Connection;
diff --git a/src/native.rs b/src/native.rs
index e3733955..e517b05d 100644
--- a/src/native.rs
+++ b/src/native.rs
@@ -1,6 +1,6 @@
 pub use iced_winit::{
-    subscription, Align, Background, Color, Command, Font, HorizontalAlignment,
-    Length, Subscription, VerticalAlignment,
+    Align, Background, Color, Command, Font, HorizontalAlignment, Length,
+    Subscription, VerticalAlignment,
 };
 
 pub mod widget {
diff --git a/winit/src/application.rs b/winit/src/application.rs
index 26ebdb05..49a01320 100644
--- a/winit/src/application.rs
+++ b/winit/src/application.rs
@@ -184,6 +184,10 @@ pub trait Application: Sized {
                 debug.layout_finished();
 
                 debug.event_processing_started();
+                events
+                    .iter()
+                    .for_each(|event| alive_subscriptions.send_event(*event));
+
                 let mut messages =
                     user_interface.update(&renderer, events.drain(..));
                 messages.extend(external_messages.drain(..));
@@ -207,7 +211,6 @@ pub trait Application: Sized {
 
                         debug.update_started();
                         let command = application.update(message);
-
                         spawn(command, &mut thread_pool, &proxy);
                         debug.update_finished();
                     }
@@ -422,7 +425,12 @@ fn spawn<Message: Send>(
 }
 
 pub struct Subscriptions {
-    alive: HashMap<u64, futures::channel::oneshot::Sender<()>>,
+    alive: HashMap<u64, Connection>,
+}
+
+pub struct Connection {
+    _cancel: futures::channel::oneshot::Sender<()>,
+    listener: Option<futures::channel::mpsc::Sender<Event>>,
 }
 
 impl Subscriptions {
@@ -440,17 +448,19 @@ impl Subscriptions {
     ) {
         use futures::{future::FutureExt, stream::StreamExt};
 
-        let handles = subscriptions.handles();
+        let connections = subscriptions.connections();
         let mut alive = std::collections::HashSet::new();
 
-        for handle in handles {
-            let id = handle.id();
+        for connection in connections {
+            let id = connection.id();
             let _ = alive.insert(id);
 
             if !self.alive.contains_key(&id) {
                 let (cancel, cancelled) = futures::channel::oneshot::channel();
+                let (event_sender, event_receiver) =
+                    futures::channel::mpsc::channel(100);
 
-                let stream = handle.stream();
+                let stream = connection.stream(event_receiver);
 
                 let proxy =
                     std::sync::Arc::new(std::sync::Mutex::new(proxy.clone()));
@@ -471,12 +481,36 @@ impl Subscriptions {
 
                 thread_pool.spawn_ok(future);
 
-                let _ = self.alive.insert(id, cancel);
+                let _ = self.alive.insert(
+                    id,
+                    Connection {
+                        _cancel: cancel,
+                        listener: if event_sender.is_closed() {
+                            None
+                        } else {
+                            Some(event_sender)
+                        },
+                    },
+                );
             }
         }
 
         self.alive.retain(|id, _| alive.contains(&id));
     }
+
+    fn send_event(&mut self, event: Event) {
+        self.alive
+            .values_mut()
+            .filter_map(|connection| connection.listener.as_mut())
+            .for_each(|listener| {
+                if let Err(error) = listener.try_send(event) {
+                    log::warn!(
+                        "Error sending event to subscription: {:?}",
+                        error
+                    );
+                }
+            });
+    }
 }
 
 // As defined in: http://www.unicode.org/faq/private_use.html