Implement ZIO#schedule (zio#4296)

* implement ZIO#schedule

* implement scheduleWith

* rename

core-tests/shared/src/test/scala/zio/ZIOSpec.scala

@@ -2957,6 +2957,18 @@ object ZIOSpec extends ZIOBaseSpec {
         io.lock(executor)
       } @@ jvm(nonFlaky(100))
     ),
+    suite("schedule")(
+      testM("runs effect for each recurrence of the schedule") {
+        for {
+          ref     <- Ref.make[List[Duration]](List.empty)
+          effect   = clock.nanoTime.flatMap(duration => ref.update(duration.nanoseconds :: _))
+          schedule = Schedule.spaced(1.second) && Schedule.recurs(5)
+          _       <- effect.schedule(schedule).fork
+          _       <- TestClock.adjust(5.seconds)
+          value   <- ref.get.map(_.reverse)
+        } yield assert(value)(equalTo(List(1.second, 2.seconds, 3.seconds, 4.seconds, 5.seconds)))
+      }
+    ),
     suite("someOrElse")(
       testM("extracts the value from Some") {
         assertM(UIO.succeed(Some(1)).someOrElse(2))(equalTo(1))

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

@@ -1665,6 +1665,27 @@ sealed trait ZIO[-R, +E, +A] extends Serializable with ZIOPlatformSpecific[R, E,
    */
   final def sandbox: ZIO[R, Cause[E], A] = foldCauseM(ZIO.fail(_), ZIO.succeedNow)
 
+  /**
+   * Runs this effect according to the specified schedule.
+   *
+   * See [[scheduleFrom]] for a variant that allows the schedule's decision to
+   * depend on the rsult of this effect.
+   */
+  final def schedule[R1 <: R, B](schedule: Schedule[R1, Any, B]): ZIO[R1 with Clock, E, B] =
+    scheduleFrom(())(schedule)
+
+  /**
+   * Runs this effect according to the specified schedule starting from the
+   * specified input value.
+   */
+  final def scheduleFrom[R1 <: R, A1 >: A, B](a: A1)(schedule: Schedule[R1, A1, B]): ZIO[R1 with Clock, E, B] =
+    schedule.driver.flatMap { driver =>
+      def loop(a: A1): ZIO[R1 with Clock, E, B] =
+        driver.next(a).foldM(_ => driver.last.orDie, _ => self.flatMap(loop))
+
+      loop(a)
+    }
+
   /**
    * Converts an option on values into an option on errors.
    */