Implement `time::every` in `iced_futures`

3 files changed, 76 insertions, 1 deletions

diff --git a/futures/Cargo.toml b/futures/Cargo.toml
index f0e2a104..275d0391 100644
--- a/futures/Cargo.toml
+++ b/futures/Cargo.toml
@@ -22,11 +22,12 @@ version = "0.3"
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies.tokio]
 version = "0.2"
 optional = true
-features = ["rt-core", "rt-threaded"]
+features = ["rt-core", "rt-threaded", "time", "stream"]
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies.async-std]
 version = "1.0"
 optional = true
+features = ["unstable"]
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 wasm-bindgen-futures = "0.4"
diff --git a/futures/src/lib.rs b/futures/src/lib.rs
index 966a9cdc..d1a149f7 100644
--- a/futures/src/lib.rs
+++ b/futures/src/lib.rs
@@ -14,6 +14,10 @@ mod runtime;
 pub mod executor;
 pub mod subscription;
 
+#[cfg(any(feature = "tokio", feature = "async-std"))]
+#[cfg_attr(docsrs, doc(cfg(any(feature = "tokio", feature = "async-std"))))]
+pub mod time;
+
 pub use command::Command;
 pub use executor::Executor;
 pub use runtime::Runtime;
diff --git a/futures/src/time.rs b/futures/src/time.rs
new file mode 100644
index 00000000..e87b4a83
--- /dev/null
+++ b/futures/src/time.rs
@@ -0,0 +1,70 @@
+//! Listen and react to time.
+use crate::subscription::{self, Subscription};
+
+/// Returns a [`Subscription`] that produces messages at a set interval.
+///
+/// The first message is produced after a `duration`, and then continues to
+/// produce more messages every `duration` after that.
+///
+/// [`Subscription`]: ../subscription/struct.Subscription.html
+pub fn every<H: std::hash::Hasher, E>(
+    duration: std::time::Duration,
+) -> Subscription<H, E, std::time::Instant> {
+    Subscription::from_recipe(Every(duration))
+}
+
+struct Every(std::time::Duration);
+
+#[cfg(feature = "async-std")]
+impl<H, E> subscription::Recipe<H, E> for Every
+where
+    H: std::hash::Hasher,
+{
+    type Output = std::time::Instant;
+
+    fn hash(&self, state: &mut H) {
+        use std::hash::Hash;
+
+        std::any::TypeId::of::<Self>().hash(state);
+        self.0.hash(state);
+    }
+
+    fn stream(
+        self: Box<Self>,
+        _input: futures::stream::BoxStream<'static, E>,
+    ) -> futures::stream::BoxStream<'static, Self::Output> {
+        use futures::stream::StreamExt;
+
+        async_std::stream::interval(self.0)
+            .map(|_| std::time::Instant::now())
+            .boxed()
+    }
+}
+
+#[cfg(all(feature = "tokio", not(feature = "async-std")))]
+impl<H, E> subscription::Recipe<H, E> for Every
+where
+    H: std::hash::Hasher,
+{
+    type Output = std::time::Instant;
+
+    fn hash(&self, state: &mut H) {
+        use std::hash::Hash;
+
+        std::any::TypeId::of::<Self>().hash(state);
+        self.0.hash(state);
+    }
+
+    fn stream(
+        self: Box<Self>,
+        _input: futures::stream::BoxStream<'static, E>,
+    ) -> futures::stream::BoxStream<'static, Self::Output> {
+        use futures::stream::StreamExt;
+
+        let start = tokio::time::Instant::now() + self.0;
+
+        tokio::time::interval_at(start, self.0)
+            .map(|_| std::time::Instant::now())
+            .boxed()
+    }
+}