1 files changed, 88 insertions, 45 deletions

diff --git a/futures/src/subscription.rs b/futures/src/subscription.rs
index b68444cd..27d2d295 100644
--- a/futures/src/subscription.rs
+++ b/futures/src/subscription.rs
@@ -3,7 +3,7 @@ mod tracker;
 
 pub use tracker::Tracker;
 
-use futures::stream::BoxStream;
+use crate::BoxStream;
 
 /// A request to listen to external events.
 ///
@@ -19,8 +19,7 @@ use futures::stream::BoxStream;
 /// This type is normally aliased by runtimes with a specific `Event` and/or
 /// `Hasher`.
 ///
-/// [`Command`]: ../struct.Command.html
-/// [`Subscription`]: struct.Subscription.html
+/// [`Command`]: crate::Command
 pub struct Subscription<Hasher, Event, Output> {
     recipes: Vec<Box<dyn Recipe<Hasher, Event, Output = Output>>>,
 }
@@ -30,8 +29,6 @@ where
     H: std::hash::Hasher,
 {
     /// Returns an empty [`Subscription`] that will not produce any output.
-    ///
-    /// [`Subscription`]: struct.Subscription.html
     pub fn none() -> Self {
         Self {
             recipes: Vec::new(),
@@ -39,9 +36,6 @@ where
     }
 
     /// Creates a [`Subscription`] from a [`Recipe`] describing it.
-    ///
-    /// [`Subscription`]: struct.Subscription.html
-    /// [`Recipe`]: trait.Recipe.html
     pub fn from_recipe(
         recipe: impl Recipe<H, E, Output = O> + 'static,
     ) -> Self {
@@ -52,8 +46,6 @@ where
 
     /// Batches all the provided subscriptions and returns the resulting
     /// [`Subscription`].
-    ///
-    /// [`Subscription`]: struct.Subscription.html
     pub fn batch(
         subscriptions: impl IntoIterator<Item = Subscription<H, E, O>>,
     ) -> Self {
@@ -66,33 +58,46 @@ where
     }
 
     /// Returns the different recipes of the [`Subscription`].
-    ///
-    /// [`Subscription`]: struct.Subscription.html
     pub fn recipes(self) -> Vec<Box<dyn Recipe<H, E, Output = O>>> {
         self.recipes
     }
 
-    /// Transforms the [`Subscription`] output with the given function.
+    /// Adds a value to the [`Subscription`] context.
     ///
-    /// [`Subscription`]: struct.Subscription.html
-    pub fn map<A>(
-        mut self,
-        f: impl Fn(O) -> A + Send + Sync + 'static,
-    ) -> Subscription<H, E, A>
+    /// The value will be part of the identity of a [`Subscription`].
+    pub fn with<T>(mut self, value: T) -> Subscription<H, E, (T, O)>
     where
         H: 'static,
         E: 'static,
         O: 'static,
-        A: 'static,
+        T: std::hash::Hash + Clone + Send + Sync + 'static,
     {
-        let function = std::sync::Arc::new(f);
+        Subscription {
+            recipes: self
+                .recipes
+                .drain(..)
+                .map(|recipe| {
+                    Box::new(With::new(recipe, value.clone()))
+                        as Box<dyn Recipe<H, E, Output = (T, O)>>
+                })
+                .collect(),
+        }
+    }
 
+    /// Transforms the [`Subscription`] output with the given function.
+    pub fn map<A>(mut self, f: fn(O) -> A) -> Subscription<H, E, A>
+    where
+        H: 'static,
+        E: 'static,
+        O: 'static,
+        A: 'static,
+    {
         Subscription {
             recipes: self
                 .recipes
                 .drain(..)
                 .map(|recipe| {
-                    Box::new(Map::new(recipe, function.clone()))
+                    Box::new(Map::new(recipe, f))
                         as Box<dyn Recipe<H, E, Output = A>>
                 })
                 .collect(),
@@ -112,22 +117,25 @@ impl<I, O, H> std::fmt::Debug for Subscription<I, O, H> {
 /// by runtimes to run and identify subscriptions. You can use it to create your
 /// own!
 ///
-/// [`Subscription`]: struct.Subscription.html
-/// [`Recipe`]: trait.Recipe.html
+/// # Examples
+/// The repository has a couple of [examples] that use a custom [`Recipe`]:
+///
+/// - [`download_progress`], a basic application that asynchronously downloads
+/// a dummy file of 100 MB and tracks the download progress.
+/// - [`stopwatch`], a watch with start/stop and reset buttons showcasing how
+/// to listen to time.
+///
+/// [examples]: https://github.com/hecrj/iced/tree/0.2/examples
+/// [`download_progress`]: https://github.com/hecrj/iced/tree/0.2/examples/download_progress
+/// [`stopwatch`]: https://github.com/hecrj/iced/tree/0.2/examples/stopwatch
 pub trait Recipe<Hasher: std::hash::Hasher, Event> {
     /// The events that will be produced by a [`Subscription`] with this
     /// [`Recipe`].
-    ///
-    /// [`Subscription`]: struct.Subscription.html
-    /// [`Recipe`]: trait.Recipe.html
     type Output;
 
     /// Hashes the [`Recipe`].
     ///
     /// This is used by runtimes to uniquely identify a [`Subscription`].
-    ///
-    /// [`Subscription`]: struct.Subscription.html
-    /// [`Recipe`]: trait.Recipe.html
     fn hash(&self, state: &mut Hasher);
 
     /// Executes the [`Recipe`] and produces the stream of events of its
@@ -135,24 +143,21 @@ pub trait Recipe<Hasher: std::hash::Hasher, Event> {
     ///
     /// 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: BoxStream<'static, Event>,
-    ) -> BoxStream<'static, Self::Output>;
+        input: BoxStream<Event>,
+    ) -> BoxStream<Self::Output>;
 }
 
 struct Map<Hasher, Event, A, B> {
     recipe: Box<dyn Recipe<Hasher, Event, Output = A>>,
-    mapper: std::sync::Arc<dyn Fn(A) -> B + Send + Sync>,
+    mapper: fn(A) -> B,
 }
 
 impl<H, E, A, B> Map<H, E, A, B> {
     fn new(
         recipe: Box<dyn Recipe<H, E, Output = A>>,
-        mapper: std::sync::Arc<dyn Fn(A) -> B + Send + Sync + 'static>,
+        mapper: fn(A) -> B,
     ) -> Self {
         Map { recipe, mapper }
     }
@@ -169,21 +174,59 @@ where
     fn hash(&self, state: &mut H) {
         use std::hash::Hash;
 
-        std::any::TypeId::of::<B>().hash(state);
         self.recipe.hash(state);
+        self.mapper.hash(state);
     }
 
-    fn stream(
-        self: Box<Self>,
-        input: BoxStream<'static, E>,
-    ) -> futures::stream::BoxStream<'static, Self::Output> {
+    fn stream(self: Box<Self>, input: BoxStream<E>) -> BoxStream<Self::Output> {
         use futures::StreamExt;
 
         let mapper = self.mapper;
 
-        self.recipe
-            .stream(input)
-            .map(move |element| mapper(element))
-            .boxed()
+        Box::pin(
+            self.recipe
+                .stream(input)
+                .map(move |element| mapper(element)),
+        )
+    }
+}
+
+struct With<Hasher, Event, A, B> {
+    recipe: Box<dyn Recipe<Hasher, Event, Output = A>>,
+    value: B,
+}
+
+impl<H, E, A, B> With<H, E, A, B> {
+    fn new(recipe: Box<dyn Recipe<H, E, Output = A>>, value: B) -> Self {
+        With { recipe, value }
+    }
+}
+
+impl<H, E, A, B> Recipe<H, E> for With<H, E, A, B>
+where
+    A: 'static,
+    B: 'static + std::hash::Hash + Clone + Send + Sync,
+    H: std::hash::Hasher,
+{
+    type Output = (B, A);
+
+    fn hash(&self, state: &mut H) {
+        use std::hash::Hash;
+
+        std::any::TypeId::of::<B>().hash(state);
+        self.value.hash(state);
+        self.recipe.hash(state);
+    }
+
+    fn stream(self: Box<Self>, input: BoxStream<E>) -> BoxStream<Self::Output> {
+        use futures::StreamExt;
+
+        let value = self.value;
+
+        Box::pin(
+            self.recipe
+                .stream(input)
+                .map(move |element| (value.clone(), element)),
+        )
     }
 }