Merge pull request #2684 from eugeniasimich/spaced

Added `spaced` function, similar to `metered` but waiting

Author: mpilquist

core/shared/src/main/scala/fs2/Stream.scala

@@ -675,6 +675,30 @@ final class Stream[+F[_], +O] private[fs2] (private[fs2] val underlying: Pull[F,
   def meteredStartImmediately[F2[x] >: F[x]: Temporal](rate: FiniteDuration): Stream[F2, O] =
     (Stream.emit(()) ++ Stream.fixedRate[F2](rate)).zipRight(this)
 
+  /** Waits the specified `delay` between each event.
+    *
+    * The resulting stream emits the same elements from `this` stream,
+    * but split into singleton chunks. Between each chunk (element) it
+    * adds a pause of a fixed `delay` duration.
+    *
+    * This method differs in the timing of elements from [[metered]].
+    * The [[metered]] combinator takes a "schedule" for elements to be released,
+    * and before each element introduces just the necessary delay to hit that time.
+    * To do so, it deducts from the pause any delay caused by other effects
+    * in the stream, or the pauses the stream consumer takes while pulling.
+    * This method, instead, simply introduced a fixed sleep time between elements,
+    * irrespective of other pauses in the stream or the consumer.
+    *
+    * Starts immediately, same as [[meteredStartImmediately]]
+    * unless parameter `startImmediately` is set to false.
+    */
+  def spaced[F2[x] >: F[x]: Temporal](
+      delay: FiniteDuration,
+      startImmediately: Boolean = true
+  ): Stream[F2, O] =
+    ((if (startImmediately) Stream.emit(()) else Stream.empty) ++ Stream.fixedDelay[F2](delay))
+      .zipRight(this)
+
   /** Logs the elements of this stream as they are pulled.
     *
     * By default, `toString` is called on each element and the result is printed

core/shared/src/test/scala/fs2/StreamCombinatorsSuite.scala

@@ -1056,6 +1056,50 @@ class StreamCombinatorsSuite extends Fs2Suite {
       .map(results => assert(results.size == 1))
   }
 
+  test("spaced should start immediately if startImmediately is not set") {
+    Stream
+      .emit[IO, Int](1)
+      .repeatN(10)
+      .spaced(1.second)
+      .interruptAfter(500.milliseconds)
+      .compile
+      .toList
+      .map(results => assert(results.size == 1))
+  }
+
+  test("spaced should not start immediately if startImmediately is set to false") {
+    Stream
+      .emit[IO, Int](1)
+      .repeatN(10)
+      .spaced(1.second, startImmediately = false)
+      .interruptAfter(500.milliseconds)
+      .compile
+      .toList
+      .map(results => assert(results.isEmpty))
+  }
+
+  test("metered should not wait between events that last longer than the rate") {
+    Stream
+      .eval[IO, Int](IO.sleep(1.second).as(1))
+      .repeatN(10)
+      .metered(1.second)
+      .interruptAfter(4500.milliseconds)
+      .compile
+      .toList
+      .map(results => assert(results.size == 3))
+  }
+
+  test("spaced should wait between events") {
+    Stream
+      .eval[IO, Int](IO.sleep(1.second).as(1))
+      .repeatN(10)
+      .spaced(1.second)
+      .interruptAfter(4500.milliseconds)
+      .compile
+      .toList
+      .map(results => assert(results.size == 2))
+  }
+
   test("mapAsyncUnordered") {
     forAllF { (s: Stream[Pure, Int]) =>
       val f = (_: Int) + 1