feat: add interval_stream helper

Author: OliverNChalk

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "toolbox"
-version = "0.1.0"
+version = "0.2.0"
 authors = ["Oliver Chalk"]
 edition = "2021"
 readme = "README.md"
@@ -14,13 +14,15 @@ disallowed_methods = "warn"
 
 [features]
 default = []
-tokio = ["dep:tokio"]
+named_task = ["tokio/rt"]
+interval_stream = ["tokio/time", "dep:futures"]
 tracing = ["dep:const_format", "dep:tracing", "dep:tracing-appender", "dep:tracing-subscriber"]
 version = ["dep:const_format"]
 
 [dependencies]
 const_format = { version = "0.2.32", optional = true }
-tokio = { version = "1.0", features = ["rt"], optional = true }
+futures = { version = "0.3.31", optional = true }
+tokio = { version = "1.0", optional = true }
 tracing = { version = "0.1.40", optional = true }
 tracing-appender = { version = "0.2.3", optional = true }
 tracing-subscriber = { version = "0.3.18", features = ["env-filter", "json"], optional = true }

Cargo.toml

@@ -1,5 +1,5 @@
 pub mod fs;
-#[cfg(feature = "tokio")]
+#[cfg(any(feature = "named_task", feature = "interval_stream"))]
 pub mod tokio;
 #[cfg(feature = "tracing")]
 pub mod tracing;

src/lib.rs

@@ -0,0 +1,67 @@
+use std::future::Future;
+use std::pin::pin;
+use std::task::Poll;
+
+use futures::{ready, Stream};
+
+pub struct IntervalStream<Poll, Fut>
+where
+    Poll: Fn() -> Fut + Unpin,
+    Fut: Unpin,
+{
+    interval: tokio::time::Interval,
+    poll: Poll,
+
+    in_progress: Option<Fut>,
+}
+
+impl<Poll, Fut, Output> IntervalStream<Poll, Fut>
+where
+    Poll: Fn() -> Fut + Unpin,
+    Fut: Future<Output = Output> + Unpin,
+{
+    pub fn new(interval: tokio::time::Interval, poll: Poll) -> Self {
+        IntervalStream { interval, poll, in_progress: None }
+    }
+}
+
+impl<PollFn, Fut, Output> Stream for IntervalStream<PollFn, Fut>
+where
+    PollFn: Fn() -> Fut + Unpin,
+    Fut: Future<Output = Output> + Unpin,
+{
+    type Item = Output;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        let this = self.get_mut();
+
+        // Poll the current future if one already exists.
+        if let Some(fut) = &mut this.in_progress {
+            let output = ready!(pin!(fut).poll(cx));
+            this.in_progress = None;
+
+            return Poll::Ready(Some(output));
+        }
+
+        // Poll the interval to see if we should create a new future.
+        ready!(this.interval.poll_tick(cx));
+
+        // Create a new future.
+        let mut fut = (this.poll)();
+
+        // Poll the future.
+        let pinned = pin!(&mut fut);
+        let poll = pinned.poll(cx);
+        match poll {
+            Poll::Ready(output) => Poll::Ready(Some(output)),
+            Poll::Pending => {
+                this.in_progress = Some(fut);
+
+                Poll::Pending
+            }
+        }
+    }
+}

src/tokio/interval_stream.rs

@@ -0,0 +1,9 @@
+#[cfg(feature = "interval_stream")]
+mod interval_stream;
+#[cfg(feature = "named_task")]
+mod named_task;
+
+#[cfg(feature = "interval_stream")]
+pub use interval_stream::*;
+#[cfg(feature = "named_task")]
+pub use named_task::*;