Merge pull request #164 from hecrj/feature/custom-runtime

Custom futures executor with `iced_futures`

2 files changed, 174 insertions, 20 deletions

diff --git a/core/src/subscription.rs b/futures/src/subscription.rs
index d9e7e388..b68444cd 100644
--- a/core/src/subscription.rs
+++ b/futures/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.
 ///
@@ -11,16 +16,16 @@
 /// For instance, you can use a [`Subscription`] to listen to a WebSocket
 /// connection, keyboard presses, mouse events, time ticks, etc.
 ///
-/// This type is normally aliased by runtimes with a specific `Input` and/or
+/// This type is normally aliased by runtimes with a specific `Event` and/or
 /// `Hasher`.
 ///
 /// [`Command`]: ../struct.Command.html
 /// [`Subscription`]: struct.Subscription.html
-pub struct Subscription<Hasher, Input, Output> {
-    recipes: Vec<Box<dyn Recipe<Hasher, Input, Output = Output>>>,
+pub struct Subscription<Hasher, Event, Output> {
+    recipes: Vec<Box<dyn Recipe<Hasher, Event, Output = Output>>>,
 }
 
-impl<H, I, O> Subscription<H, I, O>
+impl<H, E, O> Subscription<H, E, O>
 where
     H: std::hash::Hasher,
 {
@@ -38,7 +43,7 @@ where
     /// [`Subscription`]: struct.Subscription.html
     /// [`Recipe`]: trait.Recipe.html
     pub fn from_recipe(
-        recipe: impl Recipe<H, I, Output = O> + 'static,
+        recipe: impl Recipe<H, E, Output = O> + 'static,
     ) -> Self {
         Self {
             recipes: vec![Box::new(recipe)],
@@ -50,7 +55,7 @@ where
     ///
     /// [`Subscription`]: struct.Subscription.html
     pub fn batch(
-        subscriptions: impl IntoIterator<Item = Subscription<H, I, O>>,
+        subscriptions: impl IntoIterator<Item = Subscription<H, E, O>>,
     ) -> Self {
         Self {
             recipes: subscriptions
@@ -63,7 +68,7 @@ where
     /// Returns the different recipes of the [`Subscription`].
     ///
     /// [`Subscription`]: struct.Subscription.html
-    pub fn recipes(self) -> Vec<Box<dyn Recipe<H, I, Output = O>>> {
+    pub fn recipes(self) -> Vec<Box<dyn Recipe<H, E, Output = O>>> {
         self.recipes
     }
 
@@ -73,10 +78,10 @@ where
     pub fn map<A>(
         mut self,
         f: impl Fn(O) -> A + Send + Sync + 'static,
-    ) -> Subscription<H, I, A>
+    ) -> Subscription<H, E, A>
     where
         H: 'static,
-        I: 'static,
+        E: 'static,
         O: 'static,
         A: 'static,
     {
@@ -88,7 +93,7 @@ where
                 .drain(..)
                 .map(|recipe| {
                     Box::new(Map::new(recipe, function.clone()))
-                        as Box<dyn Recipe<H, I, Output = A>>
+                        as Box<dyn Recipe<H, E, Output = A>>
                 })
                 .collect(),
         }
@@ -109,7 +114,7 @@ impl<I, O, H> std::fmt::Debug for Subscription<I, O, H> {
 ///
 /// [`Subscription`]: struct.Subscription.html
 /// [`Recipe`]: trait.Recipe.html
-pub trait Recipe<Hasher: std::hash::Hasher, Input> {
+pub trait Recipe<Hasher: std::hash::Hasher, Event> {
     /// The events that will be produced by a [`Subscription`] with this
     /// [`Recipe`].
     ///
@@ -128,31 +133,32 @@ pub trait Recipe<Hasher: std::hash::Hasher, Input> {
     /// Executes the [`Recipe`] and produces the stream of events of its
     /// [`Subscription`].
     ///
-    /// It receives some generic `Input`, which is normally defined by runtimes.
+    /// It receives some stream of generic events, which is normally defined by
+    /// shells.
     ///
     /// [`Subscription`]: struct.Subscription.html
     /// [`Recipe`]: trait.Recipe.html
     fn stream(
         self: Box<Self>,
-        input: Input,
-    ) -> futures::stream::BoxStream<'static, Self::Output>;
+        input: BoxStream<'static, Event>,
+    ) -> BoxStream<'static, Self::Output>;
 }
 
-struct Map<Hasher, Input, A, B> {
-    recipe: Box<dyn Recipe<Hasher, Input, Output = A>>,
+struct Map<Hasher, Event, A, B> {
+    recipe: Box<dyn Recipe<Hasher, Event, Output = A>>,
     mapper: std::sync::Arc<dyn Fn(A) -> B + Send + Sync>,
 }
 
-impl<H, I, A, B> Map<H, I, A, B> {
+impl<H, E, A, B> Map<H, E, A, B> {
     fn new(
-        recipe: Box<dyn Recipe<H, I, Output = A>>,
+        recipe: Box<dyn Recipe<H, E, Output = A>>,
         mapper: std::sync::Arc<dyn Fn(A) -> B + Send + Sync + 'static>,
     ) -> Self {
         Map { recipe, mapper }
     }
 }
 
-impl<H, I, A, B> Recipe<H, I> for Map<H, I, A, B>
+impl<H, E, A, B> Recipe<H, E> for Map<H, E, A, B>
 where
     A: 'static,
     B: 'static,
@@ -169,7 +175,7 @@ where
 
     fn stream(
         self: Box<Self>,
-        input: I,
+        input: BoxStream<'static, E>,
     ) -> futures::stream::BoxStream<'static, Self::Output> {
         use futures::StreamExt;
 
diff --git a/futures/src/subscription/tracker.rs b/futures/src/subscription/tracker.rs
new file mode 100644
index 00000000..c8a1ee18
--- /dev/null
+++ b/futures/src/subscription/tracker.rs
@@ -0,0 +1,148 @@
+use crate::Subscription;
+
+use futures::{future::BoxFuture, sink::Sink};
+use std::collections::HashMap;
+use std::marker::PhantomData;
+
+/// A registry of subscription streams.
+///
+/// If you have an application that continuously returns a [`Subscription`],
+/// you can use a [`Tracker`] to keep track of the different recipes and keep
+/// its executions alive.
+#[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,
+{
+    /// Creates a new empty [`Tracker`].
+    ///
+    /// [`Tracker`]: struct.Tracker.html
+    pub fn new() -> Self {
+        Self {
+            subscriptions: HashMap::new(),
+            _hasher: PhantomData,
+        }
+    }
+
+    /// Updates the [`Tracker`] with the given [`Subscription`].
+    ///
+    /// A [`Subscription`] can cause new streams to be spawned or old streams
+    /// to be closed.
+    ///
+    /// The [`Tracker`] keeps track of these streams between calls to this
+    /// method:
+    ///
+    /// - If the provided [`Subscription`] contains a new [`Recipe`] that is
+    /// currently not being run, it will spawn a new stream and keep it alive.
+    /// - On the other hand, if a [`Recipe`] is currently in execution and the
+    /// provided [`Subscription`] does not contain it anymore, then the
+    /// [`Tracker`] will close and drop the relevant stream.
+    ///
+    /// It returns a list of futures that need to be spawned to materialize
+    /// the [`Tracker`] changes.
+    ///
+    /// [`Tracker`]: struct.Tracker.html
+    /// [`Subscription`]: struct.Subscription.html
+    /// [`Recipe`]: trait.Recipe.html
+    pub fn update<Message, Receiver>(
+        &mut self,
+        subscription: Subscription<Hasher, Event, Message>,
+        receiver: Receiver,
+    ) -> Vec<BoxFuture<'static, ()>>
+    where
+        Message: 'static + Send,
+        Receiver: '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(receiver.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
+    }
+
+    /// Broadcasts an event to the subscriptions currently alive.
+    ///
+    /// A subscription's [`Recipe::stream`] always receives a stream of events
+    /// as input. This stream can be used by some subscription to listen to
+    /// shell events.
+    ///
+    /// This method publishes the given event to all the subscription streams
+    /// currently open.
+    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
+                    );
+                }
+            });
+    }
+}