Add ZStream#zipWithLatest

build.sbt

@@ -109,7 +109,7 @@ lazy val streams = crossProject(JSPlatform, JVMPlatform)
   .dependsOn(core)
   .settings(stdSettings("zio-streams"))
   .settings(buildInfoSettings)
-  .settings(replSettings)
+  .settings(streamReplSettings)
   .enablePlugins(BuildInfoPlugin)
 
 lazy val streamsJVM = streams.jvm
@@ -121,7 +121,7 @@ lazy val streamsTests = crossProject(JSPlatform, JVMPlatform)
   .dependsOn(coreTests % "test->test;compile->compile")
   .settings(stdSettings("zio-streams-tests"))
   .settings(buildInfoSettings)
-  .settings(replSettings)
+  .settings(streamReplSettings)
   .enablePlugins(BuildInfoPlugin)
 
 lazy val streamsTestsJVM = streamsTests.jvm.dependsOn(coreTestsJVM % "test->compile")

core/shared/src/main/scala/zio/ZSchedule.scala

@@ -547,8 +547,7 @@ trait ZSchedule[-R, -A, +B] extends Serializable { self =>
 object ZSchedule {
 
   /**
-   * A schedule that recurs forever, producing a count of inputs.
-   * Not in alphabetic order because other vals below depend on it.
+   * A schedule that recurs forever, producing a count of repeats: 0, 1, 2, ...
    */
   final val forever: Schedule[Any, Int] = unfold(0)(_ + 1)
 

project/BuildHelper.scala

@@ -58,7 +58,28 @@ object BuildHelper {
     }
   )
 
-  val replSettings = Seq(
+  val replSettings = makeReplSettings {
+    """|import zio._
+       |import zio.console._
+       |import zio.duration._
+       |object replRTS extends DefaultRuntime {}
+       |import replRTS._
+       |implicit class RunSyntax[R >: replRTS.Environment, E, A](io: ZIO[R, E, A]){ def unsafeRun: A = replRTS.unsafeRun(io) }
+    """.stripMargin
+  }
+
+  val streamReplSettings = makeReplSettings {
+    """|import zio._
+       |import zio.console._
+       |import zio.duration._
+       |import zio.stream._
+       |object replRTS extends DefaultRuntime {}
+       |import replRTS._
+       |implicit class RunSyntax[R >: replRTS.Environment, E, A](io: ZIO[R, E, A]){ def unsafeRun: A = replRTS.unsafeRun(io) }
+    """.stripMargin
+  }
+
+  def makeReplSettings(initialCommandsStr: String) = Seq(
     // In the repl most warnings are useless or worse.
     // This is intentionally := as it's more direct to enumerate the few
     // options we do want than to try to subtract off the ones we don't.
@@ -73,14 +94,7 @@ object BuildHelper {
       "-Xsource:2.13",
       "-Yrepl-class-based"
     ),
-    initialCommands in Compile in console := """
-                                               |import zio._
-                                               |import zio.console._
-                                               |import zio.duration._
-                                               |object replRTS extends DefaultRuntime {}
-                                               |import replRTS._
-                                               |implicit class RunSyntax[R >: replRTS.Environment, E, A](io: ZIO[R, E, A]){ def unsafeRun: A = replRTS.unsafeRun(io) }
-    """.stripMargin
+    initialCommands in Compile in console := initialCommandsStr
   )
 
   def extraOptions(scalaVersion: String, optimize: Boolean) =

streams-tests/jvm/src/test/scala/zio/stream/StreamSpec.scala

@@ -184,9 +184,11 @@ class ZStreamSpec(implicit ee: org.specs2.concurrent.ExecutionEnv) extends TestR
     repeat                  $repeat
     short circuits          $repeatShortCircuits
 
-  Stream.spaced
-    spaced                        $spaced
-    spacedEither                  $spacedEither
+  Stream.schedule
+    scheduleElementsWith          $scheduleElementsWith
+    scheduleElementsEither        $scheduleElementsEither
+    scheduleWith                  $scheduleWith
+    scheduleEither                $scheduleEither
     repeated and spaced           $repeatedAndSpaced
     short circuits in schedule    $spacedShortCircuitsWhileInSchedule
     short circuits after schedule $spacedShortCircuitsAfterScheduleFinished
@@ -236,6 +238,7 @@ class ZStreamSpec(implicit ee: org.specs2.concurrent.ExecutionEnv) extends TestR
     zipWithIndex                $zipWithIndex
     zipWith ignore RHS          $zipWithIgnoreRhs
     zipWith prioritizes failure $zipWithPrioritizesFailure
+    zipWithLatest               $zipWithLatest
   """
 
   def aggregate = unsafeRun {
@@ -1470,34 +1473,50 @@ class ZStreamSpec(implicit ee: org.specs2.concurrent.ExecutionEnv) extends TestR
       } yield result must_=== List(1, 1)
     )
 
-  private def spaced =
+  private def scheduleWith =
     unsafeRun(
       Stream("A", "B", "C")
-        .spaced(Schedule.recurs(0) *> Schedule.fromFunction((_) => "!"))
+        .scheduleWith(Schedule.recurs(3) *> Schedule.fromFunction((_) => "!"))(_.toLowerCase, identity)
         .run(Sink.collectAll[String])
-        .map(_ must_=== List("A", "!", "B", "!", "C", "!"))
+        .map(_ must_=== List("a", "b", "c", "!"))
     )
 
-  private def spacedEither =
+  private def scheduleEither =
     unsafeRun(
       Stream("A", "B", "C")
-        .spacedEither(Schedule.recurs(0) *> Schedule.fromFunction((_) => 123))
+        .scheduleEither(Schedule.recurs(3) *> Schedule.fromFunction((_) => "!"))
+        .run(Sink.collectAll[Either[String, String]])
+        .map(_ must_=== List(Right("A"), Right("B"), Right("C"), Left("!")))
+    )
+
+  private def scheduleElementsWith =
+    unsafeRun(
+      Stream("A", "B", "C")
+        .scheduleElementsWith(Schedule.recurs(0) *> Schedule.fromFunction((_) => 123))(identity, _.toString)
+        .run(Sink.collectAll[String])
+        .map(_ must_=== List("A", "123", "B", "123", "C", "123"))
+    )
+
+  private def scheduleElementsEither =
+    unsafeRun(
+      Stream("A", "B", "C")
+        .scheduleElementsEither(Schedule.recurs(0) *> Schedule.fromFunction((_) => 123))
         .run(Sink.collectAll[Either[Int, String]])
         .map(_ must_=== List(Right("A"), Left(123), Right("B"), Left(123), Right("C"), Left(123)))
     )
 
   private def repeatedAndSpaced =
     unsafeRun(
       Stream("A", "B", "C")
-        .spaced(Schedule.recurs(1) *> Schedule.fromFunction((_) => "!"))
+        .scheduleElements(Schedule.recurs(1) >>> Schedule.fromFunction((_) => "!"))
         .run(Sink.collectAll[String])
         .map(_ must_=== List("A", "A", "!", "B", "B", "!", "C", "C", "!"))
     )
 
   private def spacedShortCircuitsAfterScheduleFinished =
     unsafeRun(
       Stream("A", "B", "C")
-        .spaced(Schedule.recurs(1) *> Schedule.fromFunction((_) => "!"))
+        .scheduleElements(Schedule.recurs(1) *> Schedule.fromFunction((_) => "!"))
         .take(3)
         .run(Sink.collectAll[String])
         .map(_ must_=== List("A", "A", "!"))
@@ -1506,7 +1525,7 @@ class ZStreamSpec(implicit ee: org.specs2.concurrent.ExecutionEnv) extends TestR
   private def spacedShortCircuitsWhileInSchedule =
     unsafeRun(
       Stream("A", "B", "C")
-        .spaced(Schedule.recurs(1) *> Schedule.fromFunction((_) => "!"))
+        .scheduleElements(Schedule.recurs(1) *> Schedule.fromFunction((_) => "!"))
         .take(4)
         .run(Sink.collectAll[String])
         .map(_ must_=== List("A", "A", "!", "B"))
@@ -1772,6 +1791,19 @@ class ZStreamSpec(implicit ee: org.specs2.concurrent.ExecutionEnv) extends TestR
       .map(_ must_=== Left("Ouch"))
   }
 
+  private def zipWithLatest = unsafeRun {
+    val s1 = Stream.iterate(0)(_ + 1).fixed(100.millis)
+    val s2 = Stream.iterate(0)(_ + 1).fixed(70.millis)
+
+    withLatch { release =>
+      s1.zipWithLatest(s2)((_, _))
+        .take(8)
+        .runCollect
+        .tap(_ => release)
+        .map(_ must_=== List(0 -> 0, 0 -> 1, 1 -> 1, 1 -> 2, 2 -> 2, 2 -> 3, 2 -> 4, 3 -> 4))
+    }
+  }
+
   private def interleave = unsafeRun {
     val s1 = Stream(2, 3)
     val s2 = Stream(5, 6, 7)

streams/shared/src/main/scala/zio/stream/Stream.scala

@@ -180,6 +180,11 @@ object Stream extends ZStreamPlatformSpecific {
    */
   final def halt[E](cause: Cause[E]): Stream[E, Nothing] = fromEffect(ZIO.halt(cause))
 
+  /**
+   * See [[ZStream.iterate]]
+   */
+  final def iterate[A](a: A)(f: A => A): ZStream[Any, Nothing, A] = Stream.unfold(a)(a => Some(a -> f(a)))
+
   /**
    * See [[ZStream.managed]]
    */