implement IntervalStream in Clock::stream_slots

ethereum-consensus/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 [features]
 default = ["serde", "async"]
 serde = ["hex", "serde_json", "serde_yaml"]
-async = ["tokio", "tokio-stream", "async-stream"]
+async = ["tokio", "tokio-stream"]
 spec-tests = ["serde", "serde_yaml"]
 ec = [
     "clap",
@@ -50,7 +50,6 @@ hex = { version = "0.4.3", optional = true }
 
 tokio = { version = "1.18.2", features = ["full"], optional = true }
 tokio-stream = { version = "0.1.8", optional = true }
-async-stream = { version = "0.3.3", optional = true }
 
 bs58 = "0.4.0"
 

ethereum-consensus/src/clock.rs

@@ -196,19 +196,25 @@ impl<T: TimeProvider + Send + Sync> Clock<T> {
 pub type SystemClock = Clock<SystemTimeProvider>;
 
 #[cfg(feature = "async")]
-use tokio_stream::Stream;
+use tokio::time::interval_at;
+#[cfg(feature = "async")]
+use tokio_stream::{wrappers::IntervalStream, Stream, StreamExt};
 
 #[cfg(feature = "async")]
 impl<T: TimeProvider + Send + Sync> Clock<T> {
+    /// Returns a stream of slots. The stream will start from the next slot i.e it will not emit the
+    /// current slot.
     pub fn stream_slots(&self) -> impl Stream<Item = Slot> + '_ {
-        async_stream::stream! {
-            loop {
-                let slot = self.current_slot().expect("after genesis");
-                yield slot;
-                let duration_until_next_slot = self.duration_until_slot(slot + 1);
-                tokio::time::sleep(duration_until_next_slot).await;
-            }
-        }
+        let slot = self.current_slot().expect("after genesis");
+        let duration_until_next_slot = self.duration_until_slot(slot + 1);
+
+        // Stream will not emit the `current_slot`, it starts from the next slot. This is until a custom stream implementation is added to fix this. Ref: https://github.com/ralexstokes/ethereum-consensus/pull/281#discussion_r1350932968
+        let interval = interval_at(
+            tokio::time::Instant::now() + duration_until_next_slot,
+            Duration::from_secs(self.seconds_per_slot),
+        );
+        let interval_stream = IntervalStream::new(interval);
+        interval_stream.map(move |_| self.current_slot().expect("after genesis"))
     }
 }
 
@@ -329,6 +335,7 @@ mod tests {
             }
             slots.push(slot);
         }
-        assert_eq!(slots, (current_slot..target_slot).collect::<Vec<_>>());
+        // `current_slot + 1` because the tick for the current slot is not emitted as a compromise to use Interval, IntervalStream Ref: https://github.com/ralexstokes/ethereum-consensus/pull/281#discussion_r1350932968. A custom stream implementation is required to emit the current slot.
+        assert_eq!(slots, (current_slot + 1..target_slot).collect::<Vec<_>>());
     }
 }