Add ZIO#cached (#1361)

* add ZIO#memoizeTimed

* address review feedback

* address review comments

core-tests/jvm/src/test/scala/zio/IOSpec.scala

@@ -7,6 +7,7 @@ import org.specs2.execute.Result
 import scala.collection.mutable
 import scala.util.Try
 import zio.Cause.{ die, fail, interrupt, Both }
+import zio.duration._
 
 class IOSpec(implicit ee: org.specs2.concurrent.ExecutionEnv) extends TestRuntime with GenIO with ScalaCheck {
   import Prop.forAll
@@ -45,6 +46,7 @@ class IOSpec(implicit ee: org.specs2.concurrent.ExecutionEnv) extends TestRuntim
    Check `absolve` method on IO[E, Either[E, A]] returns the same IO[E, Either[E, String]] as `IO.absolve` does. $testAbsolve
    Check non-`memoize`d IO[E, A] returns new instances on repeated calls due to referential transparency. $testNonMemoizationRT
    Check `memoize` method on IO[E, A] returns the same instance on repeated calls. $testMemoization
+   Check `cached` method on IO[E, A] returns new instances after duration. $testCached
    Check `raceAll` method returns the same IO[E, A] as `IO.raceAll` does. $testRaceAll
    Check `firstSuccessOf` method returns the same IO[E, A] as `IO.firstSuccessOf` does. $testfirstSuccessOf
    Check `zipPar` method does not swallow exit causes of loser. $testZipParInterupt
@@ -305,6 +307,19 @@ class IOSpec(implicit ee: org.specs2.concurrent.ExecutionEnv) extends TestRuntim
     )
   }
 
+  def testCached = flaky {
+    def incrementAndGet(ref: Ref[Int]): UIO[Int] = ref.update(_ + 1)
+    for {
+      ref   <- Ref.make(0)
+      cache <- incrementAndGet(ref).cached(100.milliseconds)
+      a     <- cache
+      b     <- cache
+      _     <- clock.sleep(100.milliseconds)
+      c     <- cache
+      d     <- cache
+    } yield (a must_=== b) and (b must_!== c) and (c must_=== d)
+  }
+
   def testRaceAll = {
     val io  = IO.effectTotal("supercalifragilisticexpialadocious")
     val ios = List.empty[UIO[String]]

core-tests/jvm/src/test/scala/zio/RTSSpec.scala

@@ -1473,13 +1473,4 @@ class RTSSpec(implicit ee: ExecutionEnv) extends TestRuntime with org.specs2.mat
     unsafeRun(
       IO.mergeAll(List.empty[UIO[Int]])(0)(_ + _)
     ) must_=== 0
-
-  def nonFlaky(v: => ZIO[Environment, Any, org.specs2.matcher.MatchResult[Any]]): org.specs2.matcher.MatchResult[Any] =
-    (1 to 100).foldLeft[org.specs2.matcher.MatchResult[Any]](true must_=== true) {
-      case (acc, _) =>
-        acc and unsafeRun(v)
-    }
-
-  def flaky(v: => ZIO[Environment, Any, org.specs2.matcher.MatchResult[Any]]): org.specs2.matcher.MatchResult[Any] =
-    eventually(unsafeRun(v.timeout(1.second)).get)
 }

core-tests/jvm/src/test/scala/zio/TestRuntime.scala

@@ -6,7 +6,9 @@ import org.specs2.matcher.Expectations
 import org.specs2.matcher.TerminationMatchers.terminate
 import org.specs2.specification.{ Around, AroundEach, AroundTimeout }
 
-import scala.concurrent.duration._
+import scala.concurrent.duration.{ Duration => SDuration, MICROSECONDS }
+
+import zio.duration._
 
 abstract class TestRuntime(implicit ee: org.specs2.concurrent.ExecutionEnv)
     extends BaseCrossPlatformSpec
@@ -19,15 +21,26 @@ abstract class TestRuntime(implicit ee: org.specs2.concurrent.ExecutionEnv)
       case other                                 => other
     }
 
-  override final def aroundTimeout(to: Duration)(implicit ee: ExecutionEnv): Around =
+  override final def aroundTimeout(to: SDuration)(implicit ee: ExecutionEnv): Around =
     new Around {
       def around[T: AsResult](t: => T): Result = {
         lazy val result = t
-        val termination = terminate(retries = 1000, sleep = (to.toMicros / 1000).micros)
+        val termination = terminate(retries = 1000, sleep = SDuration(to.toMicros / 1000, MICROSECONDS))
           .orSkip(_ => "TIMEOUT: " + to)(Expectations.createExpectable(result))
 
         if (!termination.toResult.isSkipped) AsResult(result)
         else termination.toResult
       }
     }
+
+  final def flaky(
+    v: => ZIO[Environment, Any, org.specs2.matcher.MatchResult[Any]]
+  ): org.specs2.matcher.MatchResult[Any] =
+    eventually(unsafeRun(v.timeout(1.second)).get)
+
+  def nonFlaky(v: => ZIO[Environment, Any, org.specs2.matcher.MatchResult[Any]]): org.specs2.matcher.MatchResult[Any] =
+    (1 to 100).foldLeft[org.specs2.matcher.MatchResult[Any]](true must_=== true) {
+      case (acc, _) =>
+        acc and unsafeRun(v)
+    }
 }

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

@@ -157,6 +157,30 @@ sealed trait ZIO[-R, +E, +A] extends Serializable { self =>
         }
     )(use)
 
+  /**
+   * Returns an effect that, if evaluated, will return the cached result of
+   * this effect. Cached results will expire after `timeToLive` duration.
+   */
+  final def cached(timeToLive: Duration): ZIO[R with Clock, Nothing, IO[E, A]] = {
+
+    def get(cache: RefM[Option[Promise[E, A]]]): ZIO[R with Clock, E, A] =
+      ZIO.uninterruptibleMask { restore =>
+        cache.updateSome {
+          case None =>
+            for {
+              p <- Promise.make[E, A]
+              _ <- self.to(p)
+              _ <- p.await.delay(timeToLive).flatMap(_ => cache.set(None)).fork
+            } yield Some(p)
+        }.flatMap(a => restore(a.get.await))
+      }
+
+    for {
+      r     <- ZIO.environment[R with Clock]
+      cache <- RefM.make[Option[Promise[E, A]]](None)
+    } yield get(cache).provide(r)
+  }
+
   /**
    * Recovers from all errors.
    *