Implement `subscription::Tracker` in `iced_core`

4 files changed, 123 insertions, 5 deletions

diff --git a/core/Cargo.toml b/core/Cargo.toml
index 0a8fd8ef..4e019ba9 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -11,7 +11,8 @@ repository = "https://github.com/hecrj/iced"
 # Exposes a future-based `Command` type
 command = ["futures"]
 # Exposes a future-based `Subscription` type
-subscription = ["futures"]
+subscription = ["futures", "log"]
 
 [dependencies]
 futures = { version = "0.3", optional = true }
+log = { version = "0.4", optional = true }
diff --git a/core/src/lib.rs b/core/src/lib.rs
index 821b09c1..6f13c310 100644
--- a/core/src/lib.rs
+++ b/core/src/lib.rs
@@ -9,7 +9,7 @@
 //! [Iced]: https://github.com/hecrj/iced
 //! [`iced_native`]: https://github.com/hecrj/iced/tree/master/native
 //! [`iced_web`]: https://github.com/hecrj/iced/tree/master/web
-#![deny(missing_docs)]
+//#![deny(missing_docs)]
 #![deny(missing_debug_implementations)]
 #![deny(unused_results)]
 #![deny(unsafe_code)]
diff --git a/core/src/subscription.rs b/core/src/subscription.rs
index d9e7e388..87e51e48 100644
--- a/core/src/subscription.rs
+++ b/core/src/subscription.rs
@@ -1,4 +1,9 @@
 //! Listen to external events in your application.
+mod tracker;
+
+pub use tracker::Tracker;
+
+use futures::stream::BoxStream;
 
 /// A request to listen to external events.
 ///
@@ -134,8 +139,8 @@ pub trait Recipe<Hasher: std::hash::Hasher, Input> {
     /// [`Recipe`]: trait.Recipe.html
     fn stream(
         self: Box<Self>,
-        input: Input,
-    ) -> futures::stream::BoxStream<'static, Self::Output>;
+        input: BoxStream<'static, Input>,
+    ) -> BoxStream<'static, Self::Output>;
 }
 
 struct Map<Hasher, Input, A, B> {
@@ -169,7 +174,7 @@ where
 
     fn stream(
         self: Box<Self>,
-        input: I,
+        input: BoxStream<'static, I>,
     ) -> futures::stream::BoxStream<'static, Self::Output> {
         use futures::StreamExt;
 
diff --git a/core/src/subscription/tracker.rs b/core/src/subscription/tracker.rs
new file mode 100644
index 00000000..826f60c0
--- /dev/null
+++ b/core/src/subscription/tracker.rs
@@ -0,0 +1,112 @@
+use crate::Subscription;
+
+use futures::{future::BoxFuture, sink::Sink};
+use std::collections::HashMap;
+use std::marker::PhantomData;
+
+#[derive(Debug)]
+pub struct Tracker<Hasher, Event> {
+    subscriptions: HashMap<u64, Execution<Event>>,
+    _hasher: PhantomData<Hasher>,
+}
+
+#[derive(Debug)]
+pub struct Execution<Event> {
+    _cancel: futures::channel::oneshot::Sender<()>,
+    listener: Option<futures::channel::mpsc::Sender<Event>>,
+}
+
+impl<Hasher, Event> Tracker<Hasher, Event>
+where
+    Hasher: std::hash::Hasher + Default,
+    Event: 'static + Send + Clone,
+{
+    pub fn new() -> Self {
+        Self {
+            subscriptions: HashMap::new(),
+            _hasher: PhantomData,
+        }
+    }
+
+    pub fn update<Message, S>(
+        &mut self,
+        subscription: Subscription<Hasher, Event, Message>,
+        sink: S,
+    ) -> Vec<BoxFuture<'static, ()>>
+    where
+        Message: 'static + Send,
+        S: 'static
+            + Sink<Message, Error = core::convert::Infallible>
+            + Unpin
+            + Send
+            + Clone,
+    {
+        use futures::{future::FutureExt, stream::StreamExt};
+
+        let mut futures = Vec::new();
+
+        let recipes = subscription.recipes();
+        let mut alive = std::collections::HashSet::new();
+
+        for recipe in recipes {
+            let id = {
+                let mut hasher = Hasher::default();
+                recipe.hash(&mut hasher);
+
+                hasher.finish()
+            };
+
+            let _ = alive.insert(id);
+
+            if self.subscriptions.contains_key(&id) {
+                continue;
+            }
+
+            let (cancel, cancelled) = futures::channel::oneshot::channel();
+
+            // TODO: Use bus if/when it supports async
+            let (event_sender, event_receiver) =
+                futures::channel::mpsc::channel(100);
+
+            let stream = recipe.stream(event_receiver.boxed());
+
+            let future = futures::future::select(
+                cancelled,
+                stream.map(Ok).forward(sink.clone()),
+            )
+            .map(|_| ());
+
+            let _ = self.subscriptions.insert(
+                id,
+                Execution {
+                    _cancel: cancel,
+                    listener: if event_sender.is_closed() {
+                        None
+                    } else {
+                        Some(event_sender)
+                    },
+                },
+            );
+
+            futures.push(future.boxed());
+        }
+
+        self.subscriptions.retain(|id, _| alive.contains(&id));
+
+        futures
+    }
+
+    pub fn broadcast(&mut self, event: Event) {
+        self.subscriptions
+            .values_mut()
+            .filter_map(|connection| connection.listener.as_mut())
+            .for_each(|listener| {
+                if let Err(error) = listener.try_send(event.clone()) {
+                    log::error!(
+                        "Error sending event to subscription: {:?}",
+                        error
+                    );
+                }
+            });
+    }
+}