Fixed ChainedFiniteStream/Signal

Author: makingthematrix

src/main/scala/io/github/makingthematrix/signals3/Finite.scala

@@ -52,7 +52,6 @@ object Finite {
         first.subscribe(this)
         first.onClose {
           second.subscribe(this)
-          first.unsubscribe(this)
         }
       } else {
         second.subscribe(this)
@@ -79,7 +78,6 @@ object Finite {
         first.subscribe(this)
         first.onClose {
           second.subscribe(this)
-          first.unsubscribe(this)
         }
       }
       else {
@@ -96,18 +94,19 @@ object Finite {
 
   @static final private[signals3] class ChainedFiniteSignal[V, W <: V](first: FiniteSignal[V], second: => FiniteSignal[W])
     extends Signal[V] with Finite[V] with SignalSubscriber {
+
     override protected[signals3] def onWire(): Unit =
       if (!first.isClosed) {
         first.subscribe(this)
         first.onClose {
           second.subscribe(this)
-          first.unsubscribe(this)
+          lastPromise.foreach(_.completeWith(second.last))
+          second.onClose { close() }
         }
       } else if (!second.isClosed) {
         second.subscribe(this)
-        second.onClose {
-          second.subscribe(this)
-        }
+        lastPromise.foreach(_.completeWith(second.last))
+        second.onClose {close()}
       }
 
     override protected[signals3] def onUnwire(): Unit =
@@ -120,13 +119,8 @@ object Finite {
       else current
     }
 
-    override protected[signals3] def changed(currentContext: Option[ExecutionContext]): Unit = {
-      update(computeValue, currentContext)
-      if (first.isClosed && second.isClosed && !isClosed) {
-        close()
-        lastPromise.foreach(_.completeWith(second.last))
-      }
-    }
+    override protected[signals3] def changed(currentContext: Option[ExecutionContext]): Unit =
+      if (!isClosed) update(computeValue, currentContext)
   }
 
   @static final private[signals3] class ChainedFiniteStream[E](first: FiniteStream[E], second: => FiniteStream[E])
@@ -136,25 +130,20 @@ object Finite {
         first.subscribe(this)
         first.onClose {
           second.subscribe(this)
-          first.unsubscribe(this)
+          lastPromise.foreach(_.completeWith(second.last))
+          second.onClose {close()}
         }
       } else if (!second.isClosed) {
         second.subscribe(this)
-        second.onClose {
-          second.subscribe(this)
-        }
+        lastPromise.foreach(_.completeWith(second.last))
+        second.onClose {close()}
       }
 
     override protected[signals3] def onUnwire(): Unit =
       if (!first.isClosed) first.unsubscribe(this)
       else if (!second.isClosed) second.unsubscribe(this)
 
-    override protected[signals3] def onEvent(event: E, currentContext: Option[ExecutionContext]): Unit = if (!isClosed) {
-      dispatch(event, currentContext)
-      if (first.isClosed && second.isClosed && !isClosed) {
-        close()
-        lastPromise.foreach(_.completeWith(second.last))
-      }
-    }
+    override protected[signals3] def onEvent(event: E, currentContext: Option[ExecutionContext]): Unit =
+      if (!isClosed) dispatch(event, currentContext)
   }
 }

src/main/scala/io/github/makingthematrix/signals3/TakeStream.scala

@@ -17,7 +17,7 @@ import scala.util.{Failure, Success}
  * @param take The number of events to take.
  * @tparam E The type of the events emitted by the stream.
  */
-class TakeStream[E](source: Stream[E], take: Int)
+final class TakeStream[E](source: Stream[E], take: Int)
   extends IndexedStream[E](source) with Finite[E] {
 
   override def isClosed: Boolean = super.isClosed || counter >= take

src/test/scala/io/github/makingthematrix/signals3/FiniteSpec.scala

@@ -110,11 +110,11 @@ class FiniteSpec extends munit.FunSuite {
   }
 
   test("Chain finite signals with >>> operator") {
-    val a: SourceStream[Int] = Stream()
-    val b: FiniteStream[Int] = a.take(3)
-    val c: SourceStream[Int] = Stream()
-    val d: FiniteStream[Int] = c.take(2)
-    val e: FiniteStream[Int] = b >>> d
+    val a: SourceSignal[Int] = Signal()
+    val b: FiniteSignal[Int] = a.take(3)
+    val c: SourceSignal[Int] = Signal()
+    val d: FiniteSignal[Int] = c.take(2)
+    val e: FiniteSignal[Int] = b >>> d
 
     var lastValue = 0
     e.last.foreach(lastValue = _)
@@ -132,17 +132,18 @@ class FiniteSpec extends munit.FunSuite {
     assertEquals(res.reverse, List(1, 2))
 
     a ! 3
-    waitFor(d, 3)
+    waitFor(e, 3)
     assert(b.isClosed)
-    assertEquals(res.reverse, List(1, 2, 3))
+    assertEquals(res.reverse, List(1, 2, 3, 20))
 
     a ! 4
     c ! 30
-    c ! 40
-    awaitAllTasks
-    assertEquals(res.reverse, Seq(1, 2, 3, 30, 40))
+    waitFor(e, 30)
     assert(d.isClosed)
     assert(e.isClosed)
-    assertEquals(lastValue, 40)
+    c ! 40
+    awaitAllTasks
+    assertEquals(res.reverse, Seq(1, 2, 3, 20, 30))
+    assertEquals(lastValue, 30)
   }
 }