Implement `Subscription::map` and `from_recipe`

4 files changed, 123 insertions, 58 deletions

diff --git a/core/src/subscription.rs b/core/src/subscription.rs
index 4c021d75..e9559f3c 100644
--- a/core/src/subscription.rs
+++ b/core/src/subscription.rs
@@ -1,60 +1,125 @@
 //! Generate events asynchronously for you application.
 
 /// An event subscription.
-pub struct Subscription<I, O> {
-    connections: Vec<Box<dyn Connection<Input = I, Output = O>>>,
+pub struct Subscription<Hasher, Input, Output> {
+    recipes: Vec<Box<dyn Recipe<Hasher, Input, Output = Output>>>,
 }
 
-impl<I, O> Subscription<I, O> {
+impl<H, I, O> Subscription<H, I, O>
+where
+    H: std::hash::Hasher,
+{
     pub fn none() -> Self {
         Self {
-            connections: Vec::new(),
+            recipes: Vec::new(),
+        }
+    }
+
+    pub fn from_recipe(
+        recipe: impl Recipe<H, I, Output = O> + 'static,
+    ) -> Self {
+        Self {
+            recipes: vec![Box::new(recipe)],
         }
     }
 
     pub fn batch(
-        subscriptions: impl Iterator<Item = Subscription<I, O>>,
+        subscriptions: impl Iterator<Item = Subscription<H, I, O>>,
     ) -> Self {
         Self {
-            connections: subscriptions
-                .flat_map(|subscription| subscription.connections)
+            recipes: subscriptions
+                .flat_map(|subscription| subscription.recipes)
                 .collect(),
         }
     }
 
-    pub fn connections(
-        self,
-    ) -> Vec<Box<dyn Connection<Input = I, Output = O>>> {
-        self.connections
+    pub fn recipes(self) -> Vec<Box<dyn Recipe<H, I, Output = O>>> {
+        self.recipes
     }
-}
 
-impl<I, O, T> From<T> for Subscription<I, O>
-where
-    T: Connection<Input = I, Output = O> + 'static,
-{
-    fn from(handle: T) -> Self {
-        Self {
-            connections: vec![Box::new(handle)],
+    pub fn map<A>(
+        mut self,
+        f: impl Fn(O) -> A + Send + Sync + 'static,
+    ) -> Subscription<H, I, A>
+    where
+        H: 'static,
+        I: 'static,
+        O: 'static,
+        A: 'static,
+    {
+        let function = std::sync::Arc::new(f);
+
+        Subscription {
+            recipes: self
+                .recipes
+                .drain(..)
+                .map(|recipe| {
+                    Box::new(Map::new(recipe, function.clone()))
+                        as Box<dyn Recipe<H, I, Output = A>>
+                })
+                .collect(),
         }
     }
 }
 
+impl<I, O, H> std::fmt::Debug for Subscription<I, O, H> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Subscription").finish()
+    }
+}
+
 /// The connection of an event subscription.
-pub trait Connection {
-    type Input;
+pub trait Recipe<Hasher: std::hash::Hasher, Input> {
     type Output;
 
-    fn id(&self) -> u64;
+    fn hash(&self, state: &mut Hasher);
 
     fn stream(
         &self,
-        input: Self::Input,
+        input: Input,
     ) -> futures::stream::BoxStream<'static, Self::Output>;
 }
 
-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()
+struct Map<Hasher, Input, A, B> {
+    recipe: Box<dyn Recipe<Hasher, Input, Output = A>>,
+    mapper: std::sync::Arc<dyn Fn(A) -> B + Send + Sync>,
+}
+
+impl<H, I, A, B> Map<H, I, A, B> {
+    fn new(
+        recipe: Box<dyn Recipe<H, I, 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>
+where
+    A: 'static,
+    B: 'static,
+    H: std::hash::Hasher,
+{
+    type Output = B;
+
+    fn hash(&self, state: &mut H) {
+        use std::hash::Hash;
+
+        std::any::TypeId::of::<B>().hash(state);
+        self.recipe.hash(state);
+    }
+
+    fn stream(
+        &self,
+        input: I,
+    ) -> futures::stream::BoxStream<'static, Self::Output> {
+        use futures::StreamExt;
+
+        let mapper = self.mapper.clone();
+
+        self.recipe
+            .stream(input)
+            .map(move |element| mapper(element))
+            .boxed()
     }
 }
diff --git a/examples/events.rs b/examples/events.rs
index 290aa975..10758519 100644
--- a/examples/events.rs
+++ b/examples/events.rs
@@ -51,7 +51,7 @@ impl Application for Events {
 
     fn subscriptions(&self) -> Subscription<Message> {
         if self.enabled {
-            events::all(Message::EventOccurred)
+            events::all().map(Message::EventOccurred)
         } else {
             Subscription::none()
         }
@@ -89,41 +89,33 @@ impl Application for Events {
 }
 
 mod events {
-    use std::sync::Arc;
-
-    pub fn all<Message>(
-        f: impl Fn(iced_native::Event) -> Message + 'static + Send + Sync,
-    ) -> iced::Subscription<Message>
-    where
-        Message: Send + 'static,
-    {
-        All(Arc::new(f)).into()
+    pub fn all() -> iced::Subscription<iced_native::Event> {
+        iced::Subscription::from_recipe(All)
     }
 
-    struct All<Message>(
-        Arc<dyn Fn(iced_native::Event) -> Message + Send + Sync>,
-    );
+    struct All;
 
-    impl<Message> iced_native::subscription::Connection for All<Message>
+    impl<H>
+        iced_native::subscription::Recipe<H, iced_native::subscription::Input>
+        for All
     where
-        Message: 'static,
+        H: std::hash::Hasher,
     {
-        type Input = iced_native::subscription::Input;
-        type Output = Message;
+        type Output = iced_native::Event;
 
-        fn id(&self) -> u64 {
-            0
+        fn hash(&self, state: &mut H) {
+            use std::hash::Hash;
+
+            std::any::TypeId::of::<All>().hash(state);
         }
 
         fn stream(
             &self,
             input: iced_native::subscription::Input,
-        ) -> futures::stream::BoxStream<'static, Message> {
+        ) -> futures::stream::BoxStream<'static, Self::Output> {
             use futures::StreamExt;
 
-            let function = self.0.clone();
-
-            input.map(move |event| function(event)).boxed()
+            input.boxed()
         }
     }
 }
diff --git a/native/src/subscription.rs b/native/src/subscription.rs
index 5fa026a2..4d000490 100644
--- a/native/src/subscription.rs
+++ b/native/src/subscription.rs
@@ -1,6 +1,6 @@
-use crate::Event;
+use crate::{Event, Hasher};
 
-pub type Subscription<T> = iced_core::Subscription<Input, T>;
+pub type Subscription<T> = iced_core::Subscription<Hasher, Input, T>;
 pub type Input = futures::channel::mpsc::Receiver<Event>;
 
-pub use iced_core::subscription::Connection;
+pub use iced_core::subscription::Recipe;
diff --git a/winit/src/application.rs b/winit/src/application.rs
index 49a01320..fb31c44f 100644
--- a/winit/src/application.rs
+++ b/winit/src/application.rs
@@ -2,8 +2,8 @@ use crate::{
     conversion,
     input::{keyboard, mouse},
     renderer::{Target, Windowed},
-    Cache, Command, Container, Debug, Element, Event, Length, MouseCursor,
-    Settings, Subscription, UserInterface,
+    Cache, Command, Container, Debug, Element, Event, Hasher, Length,
+    MouseCursor, Settings, Subscription, UserInterface,
 };
 use std::collections::HashMap;
 
@@ -448,11 +448,19 @@ impl Subscriptions {
     ) {
         use futures::{future::FutureExt, stream::StreamExt};
 
-        let connections = subscriptions.connections();
+        let recipes = subscriptions.recipes();
         let mut alive = std::collections::HashSet::new();
 
-        for connection in connections {
-            let id = connection.id();
+        for recipe in recipes {
+            let id = {
+                use std::hash::Hasher as _;
+
+                let mut hasher = Hasher::default();
+                recipe.hash(&mut hasher);
+
+                hasher.finish()
+            };
+
             let _ = alive.insert(id);
 
             if !self.alive.contains_key(&id) {
@@ -460,7 +468,7 @@ impl Subscriptions {
                 let (event_sender, event_receiver) =
                     futures::channel::mpsc::channel(100);
 
-                let stream = connection.stream(event_receiver);
+                let stream = recipe.stream(event_receiver);
 
                 let proxy =
                     std::sync::Arc::new(std::sync::Mutex::new(proxy.clone()));