JS-2170 FIX Memory leak when no events are emitted while clients await them

Author: Zschimmer

js7-base/jvm/src/main/scala/js7/base/stream/IncreasingNumberSync.scala

@@ -7,9 +7,9 @@ import js7.base.monixutils.MonixDeadline
 import js7.base.stream.IncreasingNumberSync.*
 import js7.base.time.ScalaTime.*
 import monix.eval.Task
+import monix.execution.CancelablePromise
 import org.jetbrains.annotations.TestOnly
 import scala.collection.mutable
-import scala.concurrent.Promise
 import scala.concurrent.duration.*
 
 /**
@@ -18,7 +18,7 @@ import scala.concurrent.duration.*
 final class IncreasingNumberSync(initial: Long, valueToString: Long => String)
 {
   // TODO Watch size of valueToPromise (for example via an inspection web service)
-  private val valueToPromise = mutable.TreeMap[Long, Promise[Unit]]()
+  private val valueToPromise = mutable.TreeMap[Long, CancelablePromise[Unit]]()
   @volatile private var _last = initial
 
   def onAdded(a: Long): Unit = {
@@ -66,7 +66,7 @@ final class IncreasingNumberSync(initial: Long, valueToString: Long => String)
         if (after < _last)
           RightTrue
         else {
-          val promise = valueToPromise.getOrElseUpdate(after, Promise())
+          val promise = valueToPromise.getOrElseUpdate(after, CancelablePromise())
           Task.fromFuture(promise.future)
             .as(Left(()))  // Check again
         }

js7-base/jvm/src/test/scala/js7/base/stream/IncreasingNumberSyncTest.scala

@@ -6,8 +6,8 @@ import js7.base.thread.Futures.implicits.*
 import js7.base.time.ScalaTime.*
 import js7.base.time.Stopwatch
 import monix.eval.Task
-import monix.execution.Scheduler
 import monix.execution.schedulers.TestScheduler
+import monix.execution.{CancelableFuture, Scheduler}
 import scala.concurrent.Future
 import scala.concurrent.duration.*
 
@@ -104,4 +104,25 @@ final class IncreasingNumberSyncTest extends OurTestSuite
       assert(sync.waitingCount == 0)
     }
   }
+
+  if (sys.runtime.maxMemory < 60_000_000)
+  "OutOfMemoryError" in {
+    import Scheduler.Implicits.global
+
+    val sync = new IncreasingNumberSync(initial = 0L, _.toString)
+    sync.onAdded(1)
+    val future: CancelableFuture[Boolean] = {
+      var future: CancelableFuture[Boolean] = CancelableFuture.never
+      for (i <- 1 to 20_000_000) {
+        future.cancel()
+        future = Task.defer(sync.whenAvailable(after = 1, until = Some(now + 99.seconds)))
+          .runToFuture
+        sleep(10.µs)
+      }
+      future
+    }
+    sync.onAdded(2)
+    future await 99.s
+    assert(sync.waitingCount == 0)
+  }
 }