Merge pull request #3125 from armanbilge/feature/signal-get-and-discrete-updates

Add `Signal#getAndDiscreteUpdates`

Author: mpilquist

core/shared/src/main/scala/fs2/concurrent/Signal.scala

@@ -24,7 +24,7 @@ package concurrent
 
 import cats.data.OptionT
 import cats.kernel.Eq
-import cats.effect.kernel.{Concurrent, Deferred, Ref}
+import cats.effect.kernel.{Concurrent, Deferred, Ref, Resource}
 import cats.effect.std.MapRef
 import cats.syntax.all._
 import cats.{Applicative, Functor, Invariant, Monad}
@@ -34,7 +34,7 @@ import scala.collection.immutable.LongMap
 /** Pure holder of a single value of type `A` that can be read in the effect `F`. */
 trait Signal[F[_], A] { outer =>
 
-  /** Returns a stream of the updates to this signal.
+  /** Returns a stream of the current value and subsequent updates to this signal.
     *
     * Even if you are pulling as fast as possible, updates that are very close together may
     * result in only the last update appearing in the stream. In general, when you pull
@@ -48,17 +48,33 @@ trait Signal[F[_], A] { outer =>
     */
   def continuous: Stream[F, A]
 
-  /** Asynchronously gets the current value of this `Signal`.
+  /** Gets the current value of this `Signal`.
     */
   def get: F[A]
 
+  /** Returns the current value of this `Signal` and a `Stream` to subscribe to
+    * subsequent updates, with the same semantics as [[discrete]]. The updates
+    * stream should be compiled at most once.
+    */
+  def getAndDiscreteUpdates(implicit F: Concurrent[F]): Resource[F, (A, Stream[F, A])] =
+    discrete.pull.uncons1
+      .flatMap(Pull.outputOption1(_))
+      .streamNoScope
+      .compile
+      .resource
+      .onlyOrError
+
   /** Returns a signal derived from this one, that drops update events that did not change the value.
     */
   def changes(implicit eqA: Eq[A]): Signal[F, A] =
     new Signal[F, A] {
       def discrete = outer.discrete.changes
       def continuous = outer.continuous
       def get = outer.get
+      override def getAndDiscreteUpdates(implicit F: Concurrent[F]) =
+        outer.getAndDiscreteUpdates.map { case (got, updates) =>
+          (got, updates.dropWhile(_ === got).changes)
+        }
     }
 
   /** Returns when the condition becomes true, semantically blocking
@@ -126,13 +142,23 @@ object Signal extends SignalInstances {
     new Signal[F, A] {
       def get: F[A] = F.pure(a)
       def continuous: Stream[Pure, A] = Stream.constant(a)
+      override def getAndDiscreteUpdates(implicit
+          ev: Concurrent[F]
+      ): Resource[F, (A, Stream[F, A])] =
+        Resource.pure((a, Stream.never(F)))
       def discrete: Stream[F, A] = Stream(a) ++ Stream.never
     }
 
   def mapped[F[_]: Functor, A, B](fa: Signal[F, A])(f: A => B): Signal[F, B] =
     new Signal[F, B] {
       def continuous: Stream[F, B] = fa.continuous.map(f)
       def discrete: Stream[F, B] = fa.discrete.map(f)
+      override def getAndDiscreteUpdates(implicit
+          ev: Concurrent[F]
+      ): Resource[F, (B, Stream[F, B])] =
+        fa.getAndDiscreteUpdates(ev).map { case (a, updates) =>
+          (f(a), updates.map(f))
+        }
       def get: F[B] = Functor[F].map(fa.get)(f)
     }
 
@@ -232,7 +258,16 @@ object SignallingRef {
 
           def continuous: Stream[F, A] = Stream.repeatEval(get)
 
-          def discrete: Stream[F, A] = {
+          def discrete: Stream[F, A] =
+            Stream.resource(getAndDiscreteUpdates).flatMap { case (a, updates) =>
+              Stream.emit(a) ++ updates
+            }
+
+          override def getAndDiscreteUpdates(implicit
+              ev: Concurrent[F]
+          ): Resource[F, (A, Stream[F, A])] =
+            getAndDiscreteUpdatesImpl
+          private[this] def getAndDiscreteUpdatesImpl = {
             def go(id: Long, lastSeen: Long): Stream[F, A] = {
               def getNext: F[(A, Long)] =
                 F.deferred[(A, Long)].flatMap { wait =>
@@ -252,9 +287,8 @@ object SignallingRef {
             def cleanup(id: Long): F[Unit] =
               state.update(s => s.copy(listeners = s.listeners - id))
 
-            Stream.eval(state.get).flatMap { state =>
-              Stream.emit(state.value) ++
-                Stream.bracket(newId)(cleanup).flatMap(go(_, state.lastUpdate))
+            Resource.eval(state.get).map { s =>
+              (s.value, Stream.bracket(newId)(cleanup).flatMap(go(_, s.lastUpdate)))
             }
           }
 
@@ -318,6 +352,11 @@ object SignallingRef {
 
     def get: F[B] = F.map(underlying.get)(a => lensGet(a))
 
+    override def getAndDiscreteUpdates(implicit ev: Concurrent[F]): Resource[F, (B, Stream[F, B])] =
+      underlying.getAndDiscreteUpdates.map { case (a, updates) =>
+        (lensGet(a), updates.map(lensGet))
+      }
+
     def set(b: B): F[Unit] = underlying.update(a => lensModify(a)(_ => b))
 
     override def getAndSet(b: B): F[B] =
@@ -369,6 +408,12 @@ object SignallingRef {
           def get: F[B] = fa.get.map(f)
           def discrete: Stream[F, B] = fa.discrete.map(f)
           def continuous: Stream[F, B] = fa.continuous.map(f)
+          override def getAndDiscreteUpdates(implicit
+              ev: Concurrent[F]
+          ): Resource[F, (B, Stream[F, B])] =
+            fa.getAndDiscreteUpdates(ev).map { case (a, updates) =>
+              (f(a), updates.map(f))
+            }
           def set(b: B): F[Unit] = fa.set(g(b))
           def access: F[(B, B => F[Boolean])] =
             fa.access.map { case (getter, setter) =>
@@ -471,7 +516,17 @@ object SignallingMapRef {
 
             def continuous: Stream[F, Option[V]] = Stream.repeatEval(get)
 
-            def discrete: Stream[F, Option[V]] = {
+            def discrete: Stream[F, Option[V]] =
+              Stream.resource(getAndDiscreteUpdates).flatMap { case (a, updates) =>
+                Stream.emit(a) ++ updates
+              }
+
+            override def getAndDiscreteUpdates(implicit
+                ev: Concurrent[F]
+            ): Resource[F, (Option[V], Stream[F, Option[V]])] =
+              getAndDiscreteUpdatesImpl
+
+            private[this] def getAndDiscreteUpdatesImpl = {
               def go(id: Long, lastSeen: Long): Stream[F, Option[V]] = {
                 def getNext: F[(Option[V], Long)] =
                   F.deferred[(Option[V], Long)].flatMap { wait =>
@@ -510,11 +565,13 @@ object SignallingMapRef {
                   }
                 }
 
-              Stream.eval(state.get).flatMap { state =>
-                Stream.emit(state.keys.get(k).flatMap(_.value)) ++
+              Resource.eval(state.get).map { state =>
+                (
+                  state.keys.get(k).flatMap(_.value),
                   Stream
                     .bracket(newId)(cleanup)
                     .flatMap(go(_, state.keys.get(k).fold(-1L)(_.lastUpdate)))
+                )
               }
             }
 
@@ -597,6 +654,15 @@ private[concurrent] trait SignalInstances extends SignalLowPriorityInstances {
           def continuous: Stream[F, B] = Stream.repeatEval(get)
 
           def get: F[B] = ff.get.ap(fa.get)
+
+          override def getAndDiscreteUpdates(implicit
+              ev: Concurrent[F]
+          ): Resource[F, (B, Stream[F, B])] = getAndDiscreteUpdatesImpl
+
+          private[this] def getAndDiscreteUpdatesImpl =
+            (ff.getAndDiscreteUpdates, fa.getAndDiscreteUpdates).mapN { case ((f, fs), (a, as)) =>
+              (f(a), nondeterministicZip(fs, as).map { case (f, a) => f(a) })
+            }
         }
     }
   }

core/shared/src/test/scala/fs2/concurrent/SignalSuite.scala

@@ -266,6 +266,23 @@ class SignalSuite extends Fs2Suite {
       .intercept[NoSuchElementException]
   }
 
+  test("hold consistent with getAndDiscreteUpdates") {
+    forAllF { (init: Int, stream: Stream[Pure, Int]) =>
+      TestControl.executeEmbed {
+        stream.evalMap(IO.sleep(1.second).as(_)).holdResource(init).use { sig =>
+          sig.getAndDiscreteUpdates.use { case (got, updates) =>
+            IO(assertEquals(got, init)) *>
+              updates
+                .interruptAfter(Long.MaxValue.nanos)
+                .compile
+                .toVector
+                .assertEquals(stream.compile.toVector)
+          }
+        }
+      }
+    }
+  }
+
   test("waitUntil") {
     val target = 5
     val expected = 1