Implement `Subscription::map` and `from_recipe`

1 files changed, 91 insertions, 26 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()
     }
 }