API: Hide removeObserver functionality

Fixes #10

Author: raquo

CHANGELOG.md

@@ -9,10 +9,10 @@ Breaking changes in **bold**.
   * A bunch of class / trait member fields are now `Try[A]` instead of `A`
   * Split `fire` into `fireValue` & `fireError`, etc.
   * By and large this does not affect existing Airstream usage, just customization by subclassing
+* **API: `Observable.removeObserver` and `Transaction.removeExternalObserver` are now private ([#10](https://github.com/raquo/Airstream/issues/10))**
+* **Build: Drop Scala 2.11 support**
 * New: SignalViewer
   * This serves as a warning about my intention to deprecate and eventually remove the entirety of the `State` type in Airstream. Its strictness has not proved useful, and yet has plenty of drawbacks. See [Laminar#37](https://github.com/raquo/Laminar/issues/37) for details. If you want to speak up against that, now is the time.
-   
-* **Build: Drop Scala 2.11 support**
 
 #### v0.3 – Sep 2018 
 

README.md

@@ -95,7 +95,7 @@ EventStream is a **lazy** observable. That means that it will not receive or pro
 
 When you add an Observer to a stream, it starts to send events to the observer from now on. Different streams could potentially have custom code in them overriding this behaviour however. We strive for obviousness.
 
-The result of calling `observable.addObserver(observer)(owner)` or `observable.foreach(onNext)(owner)` is a Subscription. To remove the observer you can call `observable.removeObserver(sameObserver)` or `subscription.kill()`. You can ignore the `owner` implicit param for now. Read about it later in the [Ownership](#ownership) section. 
+The result of calling `observable.addObserver(observer)(owner)` or `observable.foreach(onNext)(owner)` is a Subscription. To remove the observer manually, you can call `subscription.kill()`, but usually it's the `owner`'s job to do that. Hold that though for now, read about owners later in the [Ownership](#ownership) section.
 
 
 ### Laziness
@@ -138,7 +138,7 @@ For extra clarity, while the stream `rap` does depend on `qux`, `rap` itself has
 
 #### Stopping Observables
 
-Just like Observers can be added to streams, they can also be removed, e.g. with `removeObserver`. When you remove the last observer (internal or external) from a stream, it is said to be **stopped**. The same domino effect as when starting streams applies, except the `onStop` method recursively undoes everything that was done by `onStart` – instead of adding an InternalObserver to parent stream, we remove it, and if that causes the grand-parent stream to be stopped, we call its `onStop` method, and the chain continues upstream. 
+Just like Observers can be added to streams, they can also be removed, e.g. with `subscription.kill()`. When you remove the last observer (internal or external) from a stream, the stream is said to be **stopped**. The same domino effect as when starting streams applies, except the `onStop` method recursively undoes everything that was done by `onStart` – instead of adding an InternalObserver to parent stream, we remove it, and if that causes the grand-parent stream to be stopped, we call its `onStop` method, and the chain continues upstream. 
 
 When the dust settles, streams that are now without observers (internal or external) will be stopped, and those that still have observers will otherwise be untouched, except they will stop referencing the now-stopped observables in their lists of internal observers.
 

src/main/scala/com/raquo/airstream/core/Observable.scala

@@ -51,8 +51,9 @@ trait Observable[+A] {
     subscription
   }
 
-  // @TODO[Bug] See https://github.com/raquo/Airstream/issues/10
-  /** Schedule unsubscription from an external observer in the next transaction.
+  /** Note: this is core-private for subscription safety. See https://github.com/raquo/Airstream/issues/10
+    *
+    * Schedule unsubscription from an external observer in the next transaction.
     *
     * This will still happen synchronously, but will not interfere with
     * iteration over the observables' lists of observers during the current
@@ -61,7 +62,7 @@ trait Observable[+A] {
     * Note: To completely unsubscribe an Observer from this Observable,
     * you need to remove it as many times as you added it to this Observable.
     */
-  @inline def removeObserver(observer: Observer[A]): Unit = {
+  @inline private[core] def removeObserver(observer: Observer[A]): Unit = {
     Transaction.removeExternalObserver(this, observer)
   }
 

src/main/scala/com/raquo/airstream/core/Transaction.scala

@@ -35,10 +35,13 @@ object Transaction { // extends GlobalCounter {
 
   private[this] val pendingObserverRemovals: js.Array[() => Unit] = js.Array()
 
-  /** Safely remove external observer (such that it doesn't interfere with iteration over the list of observers).
-    * Removal still happens synchronously, just at the end of a transaction if one is running right now.
+  /** Note: this is core-private for subscription safety. See https://github.com/raquo/Airstream/issues/10
+    *
+    * Safely remove external observer (such that it doesn't interfere with iteration over the list of observers).
+    * Removal still happens synchronously, just at the end of a transaction if one is running right now, so that it
+    * does not interfere with iteration over the observables' lists of observers during the current transaction.
     */
-  def removeExternalObserver[A](observable: Observable[A], observer: Observer[A]): Unit = {
+  private[core] def removeExternalObserver[A](observable: Observable[A], observer: Observer[A]): Unit = {
     if (isSafeToRemoveObserver) {
       // remove right now – useful for efficient recursive removals
       observable.removeExternalObserverNow(observer)

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

@@ -4,7 +4,7 @@ import com.raquo.airstream.core.AirstreamError.ObserverError
 import com.raquo.airstream.core.{AirstreamError, LazyObservable, MemoryObservable, Transaction}
 import com.raquo.airstream.features.CombineObservable
 import com.raquo.airstream.ownership.Owner
-import com.raquo.airstream.signal.{FoldSignal, Signal, SignalFromEventStream}
+import com.raquo.airstream.signal.{FoldSignal, Signal, SignalFromEventStream, SignalViewer}
 import com.raquo.airstream.state.{MapState, State}
 
 import scala.concurrent.Future

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

@@ -77,7 +77,13 @@ trait Signal[+A] extends MemoryObservable[A] with LazyObservable[A] {
 
   override def recoverToTry: Signal[Try[A]] = map(Try(_)).recover[Try[A]] { case err => Some(Failure(err)) }
 
-  def observe(implicit owner: Owner): SignalViewer[A] = new SignalViewer[A](this, owner)
+  /** Add a noop observer to this signal to ensure that it's started.
+    * This lets you access .now and .tryNow on the resulting SignalViewer.
+    *
+    * You can use `myStream.toWeakSignal.observe` to read the last emitted
+    * value from event streams just as well.
+    */
+  def observe(implicit owner: Owner): SignalViewer[A] = new SignalViewer(this, owner)
 
   /** Initial value is only evaluated if/when needed (when there are observers) */
   override protected[airstream] def tryNow(): Try[A] = {

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

@@ -5,6 +5,11 @@ import com.raquo.airstream.ownership.Owner
 
 import scala.util.Try
 
+/** This class adds a noop observer to `signal`, ensuring that its current value is computed.
+  * It then lets you query `signal`'s current value with `now` and `tryNow` methods.
+  *
+  * This will likely replace the entirety of the `State` type in Airstream.
+  */
 class SignalViewer[+A](signal: Signal[A], owner: Owner) {
 
   private[this] val subscription: Subscription = signal.addObserver(Observer.empty)(owner)

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

@@ -40,23 +40,23 @@ class EventBusSpec extends FunSpec with Matchers {
     val obs1 = Observer[Int](newValue => effects += Effect("obs1", newValue))
     val obs2 = Observer[Int](newValue => effects += Effect("obs2", newValue))
 
-    bus.events.addObserver(obs1)
-    bus.events.addObserver(obs2)
+    val sub1 = bus.events.addObserver(obs1)
+    val sub2 = bus.events.addObserver(obs2)
     val subscription3 = bus.events.foreach(newValue => effects += Effect("obs3", newValue))
 
     bus.writer.onNext(5)
 
     effects shouldEqual mutable.Buffer(Effect("obs1", 5), Effect("obs2", 5), Effect("obs3", 5))
     effects.clear()
 
-    bus.events.removeObserver(obs2)
+    sub2.kill()
 
     bus.writer.onNext(6)
 
     effects shouldEqual mutable.Buffer(Effect("obs1", 6), Effect("obs3", 6))
     effects.clear()
 
-    bus.events.removeObserver(obs1)
+    sub1.kill()
 
     bus.writer.onNext(7)
     bus.writer.onNext(8)

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

@@ -38,7 +38,7 @@ class SampleCombineEventStream2Spec extends FunSpec with Matchers {
 
     // --
 
-    combinedStream.addObserver(combinedObserver)
+    val subCombined = combinedStream.addObserver(combinedObserver)
 
     calculations shouldEqual mutable.Buffer(
       Calculation("signal", 0)
@@ -90,7 +90,7 @@ class SampleCombineEventStream2Spec extends FunSpec with Matchers {
 
     // --
 
-    combinedStream.removeObserver(combinedObserver)
+    subCombined.kill()
     sampledSignal.addObserver(signalObserver)
 
     calculations shouldEqual mutable.Buffer()
@@ -168,7 +168,7 @@ class SampleCombineEventStream2Spec extends FunSpec with Matchers {
 
     // --
 
-    combinedStream.addObserver(combinedObserver)
+    val subCombined = combinedStream.addObserver(combinedObserver)
 
     calculations shouldEqual mutable.Buffer()
     effects shouldEqual mutable.Buffer()
@@ -218,7 +218,7 @@ class SampleCombineEventStream2Spec extends FunSpec with Matchers {
 
     // --
 
-    combinedStream.removeObserver(combinedObserver)
+    subCombined.kill()
 
     calculations shouldEqual mutable.Buffer()
     effects shouldEqual mutable.Buffer()

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

@@ -32,7 +32,7 @@ class SwitchEventStreamSpec extends FunSpec with Matchers {
     val flattenStream = $latestNumber
       .map(Calculation.log("flattened", calculations))
 
-    flattenStream.addObserver(flattenObserver)
+    val subFlatten = flattenStream.addObserver(flattenObserver)
 
     calculations shouldEqual mutable.Buffer()
     effects shouldEqual mutable.Buffer()
@@ -93,7 +93,7 @@ class SwitchEventStreamSpec extends FunSpec with Matchers {
     val sourceStream2Observer = Observer[Int](effects += Effect("source-2-obs", _))
 
     sourceStreams(2).addObserver(sourceStream2Observer)
-    flattenStream.removeObserver(flattenObserver)
+    subFlatten.kill()
 
     calculations shouldEqual mutable.Buffer()
     effects shouldEqual mutable.Buffer()

src/test/scala/com/raquo/airstream/signal/FoldSignalSpec.scala

@@ -33,7 +33,7 @@ class FoldSignalSpec extends FunSpec with Matchers {
 
     // --
 
-    signal.addObserver(signalObserver)
+    val sub = signal.addObserver(signalObserver)
 
     calculations shouldEqual mutable.Buffer(
       Calculation("signal", "numbers:")
@@ -61,7 +61,7 @@ class FoldSignalSpec extends FunSpec with Matchers {
 
     // --
 
-    signal.removeObserver(signalObserver)
+    sub.kill()
     bus.writer.onNext(3)
 
     signal.addObserver(signalObserver)

src/test/scala/com/raquo/airstream/signal/SignalSpec.scala

@@ -42,7 +42,7 @@ class SignalSpec extends FunSpec with Matchers {
     // When observer is added, it immediately gets the last evaluated current value
     // Note: Because signal had no observer when bus fired value `1`, that event was NOT used to compute new value.
     //       This is expected. If this is undesirable to your case, use State instead of Signal.
-    signal.addObserver(signalObserver1)
+    val sub1 = signal.addObserver(signalObserver1)
 
     calculations shouldEqual mutable.Buffer(
       Calculation("map-signal", -1) // First time current value of this signal was needed, so it is now calculated
@@ -86,7 +86,7 @@ class SignalSpec extends FunSpec with Matchers {
 
     // When adding a new observer, it gets the signal's current value.
     // Here the current value has been updated by the previous event, and the signal remembers it.
-    signal.addObserver(signalObserver2)
+    val sub2 = signal.addObserver(signalObserver2)
 
     calculations shouldEqual mutable.Buffer() // Using cached calculation
     effects shouldEqual mutable.Buffer(
@@ -113,7 +113,7 @@ class SignalSpec extends FunSpec with Matchers {
 
     // --
 
-    signal.removeObserver(signalObserver1)
+    sub1.kill()
 
     bus.writer.onNext(4)
 
@@ -130,7 +130,7 @@ class SignalSpec extends FunSpec with Matchers {
 
     // --
 
-    signal.removeObserver(signalObserver2)
+    sub2.kill()
 
     bus.writer.onNext(5)
 
@@ -197,7 +197,7 @@ class SignalSpec extends FunSpec with Matchers {
 
     // --
 
-    changes.addObserver(changesObserver)
+    val subChanges1 = changes.addObserver(changesObserver)
 
     bus.writer.onNext(2)
 
@@ -217,7 +217,7 @@ class SignalSpec extends FunSpec with Matchers {
     // --
 
     // Adding observer to signal sends the last evaluated current value to it
-    signal.addObserver(signalObserver)
+    val subSignal = signal.addObserver(signalObserver)
 
     calculations shouldEqual mutable.Buffer()
     effects shouldEqual mutable.Buffer(
@@ -258,7 +258,7 @@ class SignalSpec extends FunSpec with Matchers {
 
     // --
 
-    changes.removeObserver(changesObserver)
+    subChanges1.kill()
 
     bus.writer.onNext(4)
 
@@ -275,15 +275,15 @@ class SignalSpec extends FunSpec with Matchers {
 
     // --
 
-    changes.addObserver(changesObserver)
+    val subChanges2 = changes.addObserver(changesObserver)
 
     calculations shouldEqual mutable.Buffer()
     effects shouldEqual mutable.Buffer()
 
     // --
 
-    changes.removeObserver(changesObserver)
-    signal.removeObserver(signalObserver)
+    subChanges2.kill()
+    subSignal.kill()
 
     bus.writer.onNext(5)
 

src/test/scala/com/raquo/airstream/state/StateSpec.scala

@@ -75,7 +75,7 @@ class StateSpec extends FunSpec with Matchers {
 
     // --
 
-    state1.addObserver(stateObserver1)
+    val sub1 = state1.addObserver(stateObserver1)
 
     calculations shouldEqual mutable.Buffer()
     effects shouldEqual mutable.Buffer(
@@ -102,7 +102,7 @@ class StateSpec extends FunSpec with Matchers {
 
     // --
 
-    state2.addObserver(stateObserver2)
+    val sub2 = state2.addObserver(stateObserver2)
 
     calculations shouldEqual mutable.Buffer()
     effects shouldEqual mutable.Buffer(
@@ -130,8 +130,8 @@ class StateSpec extends FunSpec with Matchers {
 
     // --
 
-    state2.removeObserver(stateObserver2)
-    state1.removeObserver(stateObserver1)
+    sub2.kill()
+    sub1.kill()
 
     calculations shouldEqual mutable.Buffer()
     effects shouldEqual mutable.Buffer()