Make `subscription::Recipe` cross-platform

By removing the `Send` requirement when targetting Wasm

2 files changed, 20 insertions, 24 deletions

diff --git a/futures/src/subscription.rs b/futures/src/subscription.rs
index 8eccb7be..ab333a20 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.
 ///
@@ -168,8 +168,8 @@ pub trait Recipe<Hasher: std::hash::Hasher, Event> {
     /// [`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> {
@@ -201,18 +201,16 @@ where
         self.recipe.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)),
+        )
     }
 }
 
@@ -243,17 +241,15 @@ where
         self.recipe.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 value = self.value;
 
-        self.recipe
-            .stream(input)
-            .map(move |element| (value.clone(), element))
-            .boxed()
+        Box::pin(
+            self.recipe
+                .stream(input)
+                .map(move |element| (value.clone(), element)),
+        )
     }
 }
diff --git a/futures/src/subscription/tracker.rs b/futures/src/subscription/tracker.rs
index cfa36170..efb464b5 100644
--- a/futures/src/subscription/tracker.rs
+++ b/futures/src/subscription/tracker.rs
@@ -1,6 +1,6 @@
-use crate::Subscription;
+use crate::{BoxFuture, Subscription};
 
-use futures::{channel::mpsc, future::BoxFuture, sink::Sink};
+use futures::{channel::mpsc, sink::Sink};
 use std::{collections::HashMap, marker::PhantomData};
 
 /// A registry of subscription streams.
@@ -59,7 +59,7 @@ where
         &mut self,
         subscription: Subscription<Hasher, Event, Message>,
         receiver: Receiver,
-    ) -> Vec<BoxFuture<'static, ()>>
+    ) -> Vec<BoxFuture<()>>
     where
         Message: 'static + Send,
         Receiver: 'static
@@ -70,7 +70,7 @@ where
     {
         use futures::{future::FutureExt, stream::StreamExt};
 
-        let mut futures = Vec::new();
+        let mut futures: Vec<BoxFuture<()>> = Vec::new();
 
         let recipes = subscription.recipes();
         let mut alive = std::collections::HashSet::new();
@@ -115,7 +115,7 @@ where
                 },
             );
 
-            futures.push(future.boxed());
+            futures.push(Box::pin(future));
         }
 
         self.subscriptions.retain(|id, _| alive.contains(&id));