Fix: Avoid redundant re-starting in flattened switch observables. Fixes

#55

src/main/scala/com/raquo/airstream/eventstream/SwitchEventStream.scala

@@ -53,11 +53,16 @@ class SwitchEventStream[I, O](
   )
 
   override protected[airstream] def onNext(nextValue: I, transaction: Transaction): Unit = {
-    removeInternalObserverFromCurrentEventStream()
     val nextStream = makeStream(nextValue)
-    maybeCurrentEventStream = Success(nextStream)
-    // If we're receiving events, this stream is started, so no need to check for that
-    nextStream.addInternalObserver(internalEventObserver)
+    val isSameStream = maybeCurrentEventStream.exists { currentStream =>
+      currentStream.isSuccess && (currentStream.get eq nextStream)
+    }
+    if (!isSameStream) {
+      removeInternalObserverFromCurrentEventStream()
+      maybeCurrentEventStream = Success(nextStream)
+      // If we're receiving events, this stream is started, so no need to check for that
+      nextStream.addInternalObserver(internalEventObserver)
+    }
   }
 
   override protected[airstream] def onError(nextError: Throwable, transaction: Transaction): Unit = {

src/main/scala/com/raquo/airstream/signal/SwitchSignal.scala

@@ -32,16 +32,19 @@ class SwitchSignal[A](
   )
 
   override protected[airstream] def onTry(nextSignalTry: Try[Signal[A]], transaction: Transaction): Unit = {
-    removeInternalObserverFromCurrentSignal()
-    currentSignalTry = nextSignalTry
+    val isSameSignal = nextSignalTry.isSuccess && nextSignalTry == currentSignalTry
+    if (!isSameSignal) {
+      removeInternalObserverFromCurrentSignal()
+      currentSignalTry = nextSignalTry
 
-    // If we're receiving events, this signal is started, so no need to check for that
-    nextSignalTry.foreach { nextSignal =>
-      nextSignal.addInternalObserver(internalEventObserver)
+      // If we're receiving events, this signal is started, so no need to check for that
+      nextSignalTry.foreach { nextSignal =>
+        nextSignal.addInternalObserver(internalEventObserver)
+      }
+      //println(s"> init trx from SwitchSignal.onTry")
+      // Update this signal's value with nextSignal's current value (or an error if we don't have nextSignal)
+      new Transaction(fireTry(nextSignalTry.flatMap(_.tryNow()), _))
     }
-    //println(s"> init trx from SwitchSignal.onTry")
-    // Update this signal's value with nextSignal's current value (or an error if we don't have nextSignal)
-    new Transaction(fireTry(nextSignalTry.flatMap(_.tryNow()), _))
   }
 
   override protected[this] def onStart(): Unit = {

src/test/scala/com/raquo/airstream/eventstream/SwitchEventStreamSpec.scala

@@ -252,4 +252,294 @@ class SwitchEventStreamSpec extends UnitSpec {
     effects.clear()
   }
 
+  it("EventStream: emitting the same inner stream does not cause it to stop and re-start") {
+
+    implicit val owner: TestableOwner = new TestableOwner
+
+    val outerBus = new EventBus[Int]
+
+    val calculations = mutable.Buffer[Calculation[String]]()
+
+    // It's important that we reuse the exact same references to inner streams to check the logic
+    // - fromSeq streams are used to ensure that onStart isn't called extraneously
+    // - bus.events streams are used to ensure that onStop isn't called extraneously
+
+    val smallBus = new EventBus[String]
+
+    val smallStream = EventStream.merge(
+      smallBus.events,
+      EventStream.fromSeq("small-1" :: "small-2" :: Nil, emitOnce = true)
+    )
+
+    val bigBus = new EventBus[String]
+
+    val bigStream = EventStream.merge(
+      bigBus.events,
+      EventStream.fromSeq("big-1" :: "big-2" :: Nil, emitOnce = true)
+    )
+
+    val flatStream = outerBus.events.flatMap {
+      case i if i >= 10 => bigStream
+      case _ => smallStream
+    }.map(Calculation.log("flat", calculations))
+
+    // --
+
+    flatStream.addObserver(Observer.empty)
+
+    assert(calculations.isEmpty)
+
+    // --
+
+    outerBus.writer.onNext(1)
+
+    assert(calculations.toList == List(
+      Calculation("flat", "small-1"),
+      Calculation("flat", "small-2"),
+    ))
+
+    calculations.clear()
+
+    // --
+
+    smallBus.writer.onNext("small-bus-1")
+
+    assert(calculations.toList == List(
+      Calculation("flat", "small-bus-1")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    outerBus.writer.onNext(2)
+
+    assert(calculations.isEmpty)
+
+    // --
+
+    smallBus.writer.onNext("small-bus-2")
+
+    assert(calculations.toList == List(
+      Calculation("flat", "small-bus-2")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    outerBus.writer.onNext(10) // #Note switch to big
+
+    assert(calculations.toList == List(
+      Calculation("flat", "big-1"),
+      Calculation("flat", "big-2")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    smallBus.writer.onNext("small bus - unrelated change")
+
+    assert(calculations.isEmpty)
+
+    // --
+
+    bigBus.writer.onNext("big-bus-1")
+
+    assert(calculations.toList == List(
+      Calculation("flat", "big-bus-1")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    outerBus.writer.onNext(11)
+
+    assert(calculations.isEmpty)
+
+    // --
+
+    bigBus.writer.onNext("big-bus-2")
+
+    assert(calculations.toList == List(
+      Calculation("flat", "big-bus-2")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    outerBus.writer.onNext(5) // #Note switch back to small
+
+    assert(calculations.isEmpty) // empty because of emitOnce = true
+
+    // --
+
+    smallBus.writer.onNext("small-bus-3")
+
+    assert(calculations.toList == List(
+      Calculation("flat", "small-bus-3")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    bigBus.writer.onNext("big bus - unrelated change")
+
+    assert(calculations.isEmpty)
+  }
+
+  it("Signal: emitting the same inner stream does not cause it to stop and re-start") {
+
+    implicit val owner: TestableOwner = new TestableOwner
+
+    val outerBus = new EventBus[Int]
+
+    val calculations = mutable.Buffer[Calculation[String]]()
+
+    // It's important that we reuse the exact same references to inner streams to check the logic
+    // - fromSeq streams are used to ensure that onStart isn't called extraneously
+    // - bus.events streams are used to ensure that onStop isn't called extraneously
+
+    val smallBus = new EventBus[String]
+
+    val smallStream = EventStream.merge(
+      smallBus.events,
+      EventStream.fromSeq("small-1" :: "small-2" :: Nil, emitOnce = true)
+    )
+
+    val bigBus = new EventBus[String]
+
+    val bigStream = EventStream.merge(
+      bigBus.events,
+      EventStream.fromSeq("big-1" :: "big-2" :: Nil, emitOnce = true)
+    )
+
+    val flatStream = outerBus.events.startWith(0).flatMap {
+      case i if i >= 10 => bigStream
+      case _ => smallStream
+    }.map(Calculation.log("flat", calculations))
+
+    // --
+
+    flatStream.addObserver(Observer.empty)
+
+    assert(calculations.toList == List(
+      Calculation("flat", "small-1"),
+      Calculation("flat", "small-2"),
+    ))
+
+    calculations.clear()
+
+    // --
+
+    smallBus.writer.onNext("small-bus-0")
+
+    assert(calculations.toList == List(
+      Calculation("flat", "small-bus-0")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    outerBus.writer.onNext(1)
+
+    assert(calculations.isEmpty) // Signal == filter eats this up
+
+    // --
+
+    smallBus.writer.onNext("small-bus-1")
+
+    assert(calculations.toList == List(
+      Calculation("flat", "small-bus-1")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    outerBus.writer.onNext(2) // Signal == filter eats this up
+
+    assert(calculations.isEmpty)
+
+    // --
+
+    smallBus.writer.onNext("small-bus-2")
+
+    assert(calculations.toList == List(
+      Calculation("flat", "small-bus-2")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    outerBus.writer.onNext(10) // #Note switch to big
+
+    assert(calculations.toList == List(
+      Calculation("flat", "big-1"),
+      Calculation("flat", "big-2")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    smallBus.writer.onNext("small bus - unrelated change")
+
+    assert(calculations.isEmpty)
+
+    // --
+
+    bigBus.writer.onNext("big-bus-1")
+
+    assert(calculations.toList == List(
+      Calculation("flat", "big-bus-1")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    outerBus.writer.onNext(11)
+
+    assert(calculations.isEmpty)
+
+    // --
+
+    bigBus.writer.onNext("big-bus-2")
+
+    assert(calculations.toList == List(
+      Calculation("flat", "big-bus-2")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    outerBus.writer.onNext(5) // #Note switch back to small
+
+    assert(calculations.isEmpty) // empty because of emitOnce = true
+
+    // --
+
+    smallBus.writer.onNext("small-bus-3")
+
+    assert(calculations.toList == List(
+      Calculation("flat", "small-bus-3")
+    ))
+
+    calculations.clear()
+
+    // --
+
+    bigBus.writer.onNext("big bus - unrelated change")
+
+    assert(calculations.isEmpty)
+  }
+
 }