Issue monix#42 - fix switch, add SerialCancelable

Author: alexandru

shared/src/main/scala/monifu/concurrent/cancelables/MultiAssignmentCancelable.scala

@@ -13,30 +13,35 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
- 
+
 package monifu.concurrent.cancelables
 
 import monifu.concurrent.Cancelable
 import monifu.concurrent.atomic.Atomic
-import monifu.concurrent.cancelables.MultiAssignmentCancelable.State
-import monifu.concurrent.cancelables.MultiAssignmentCancelable.State.{Cancelled, Active}
 import scala.annotation.tailrec
 
 /**
- * Represents a [[monifu.concurrent.Cancelable]] whose underlying cancelable reference can be swapped for another.
+ * Represents a [[monifu.concurrent.Cancelable]] whose underlying cancelable
+ * reference can be swapped for another.
  *
  * Example:
  * {{{
  *   val s = MultiAssignmentCancelable()
- *   s() = c1 // sets the underlying cancelable to c1
- *   s() = c2 // swaps the underlying cancelable to c2
+ *   s := c1 // sets the underlying cancelable to c1
+ *   s := c2 // swaps the underlying cancelable to c2
  *
  *   s.cancel() // also cancels c2
  *
- *   s() = c3 // also cancels c3, because s is already canceled
+ *   s := c3 // also cancels c3, because s is already canceled
  * }}}
+ *
+ * Also see [[SerialCancelable]], which is similar, except that it cancels the
+ * old cancelable upon assigning a new cancelable.
  */
 final class MultiAssignmentCancelable private () extends BooleanCancelable {
+  import MultiAssignmentCancelable.State
+  import MultiAssignmentCancelable.State._
+
   private[this] val state = Atomic(Active(Cancelable()) : State)
 
   def isCanceled: Boolean = state.get match {
@@ -87,8 +92,8 @@ object MultiAssignmentCancelable {
     ms
   }
 
-  sealed trait State
-  object State {
+  private sealed trait State
+  private object State {
     case class Active(s: Cancelable) extends State
     case object Cancelled extends State
   }

shared/src/main/scala/monifu/concurrent/cancelables/SerialCancelable.scala

@@ -0,0 +1,104 @@
+/*
+ * Copyright (c) 2014-2015 Alexandru Nedelcu
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package monifu.concurrent.cancelables
+
+import monifu.concurrent.Cancelable
+import monifu.concurrent.atomic.Atomic
+import scala.annotation.tailrec
+
+/**
+ * Represents a [[monifu.concurrent.Cancelable]] whose underlying cancelable
+ * can be swapped for another cancelable which causes the previous underlying
+ * cancelable to be canceled.
+ *
+ * Example:
+ * {{{
+ *   val s = SerialCancelable()
+ *   s := c1 // sets the underlying cancelable to c1
+ *   s := c2 // cancels c1 and swaps the underlying cancelable to c2
+ *
+ *   s.cancel() // also cancels c2
+ *
+ *   s() = c3 // also cancels c3, because s is already canceled
+ * }}}
+ *
+ * Also see [[MultiAssignmentCancelable]], which is similar, but doesn't cancel
+ * the old cancelable upon assignment.
+ */
+final class SerialCancelable private () extends BooleanCancelable {
+  import SerialCancelable.State
+  import SerialCancelable.State._
+
+  private[this] val state = Atomic(Active(Cancelable()) : State)
+
+  def isCanceled: Boolean = state.get match {
+    case Cancelled => true
+    case _ => false
+  }
+
+  @tailrec
+  def cancel(): Boolean = state.get match {
+    case Cancelled => false
+    case current @ Active(s) =>
+      if (state.compareAndSet(current, Cancelled)) {
+        s.cancel()
+        true
+      }
+      else
+        cancel()
+  }
+
+  /**
+   * Swaps the underlying cancelable reference with `s`.
+   *
+   * In case this `SerialCancelable` is already canceled,
+   * then the reference `value` will also be canceled on assignment.
+   */
+  @tailrec
+  def update(value: Cancelable): Unit = state.get match {
+    case Cancelled => value.cancel()
+    case current @ Active(s) =>
+      if (!state.compareAndSet(current, Active(value)))
+        update(value)
+      else
+        s.cancel()
+  }
+
+  /**
+   * Alias for `update(value)`
+   */
+  def `:=`(value: Cancelable): Unit =
+    update(value)
+}
+
+object SerialCancelable {
+  def apply(): SerialCancelable =
+    new SerialCancelable()
+
+  def apply(s: Cancelable): SerialCancelable = {
+    val ms = new SerialCancelable()
+    ms() = s
+    ms
+  }
+
+  private sealed trait State
+
+  private object State {
+    case class Active(s: Cancelable) extends State
+    case object Cancelled extends State
+  }
+}

shared/src/main/scala/monifu/reactive/Observable.scala

@@ -348,6 +348,24 @@ trait Observable[+T] { self =>
       (implicit ev: T <:< Observable[U]): Observable[U] =
     operators.flatten.merge(self, bufferPolicy, delayErrors = true)
 
+  /**
+   * Convert an Observable that emits Observables into a single Observable that
+   * emits the items emitted by the most-recently-emitted of those Observables.
+   */
+  def switch[U](implicit ev: T <:< Observable[U]): Observable[U] =
+    operators.switch(self, delayErrors = false)
+
+  /**
+   * Convert an Observable that emits Observables into a single Observable that
+   * emits the items emitted by the most-recently-emitted of those Observables.
+   *
+   * It's like [[Observable.switch]], except that the created observable is
+   * reserving onError notifications until all of the Observables complete
+   * and only then passing the error along to the observers.
+   */
+  def switchDelayError[U](implicit ev: T <:< Observable[U]): Observable[U] =
+    operators.switch(self, delayErrors = true)
+
   /**
    * Given the source observable and another `Observable`, emits all of the items
    * from the first of these Observables to emit an item and cancel the other.
@@ -1168,13 +1186,6 @@ trait Observable[+T] { self =>
   def lift[U](f: Observable[T] => Observable[U]): Observable[U] =
     f(self)
 
-
-  /**
-   *
-   */
-  def switch[U](implicit ev: T <:< Observable[U]): Observable[U] =
-    operators.switch(self)
-
   /**
    * Returns the first generated result as a Future and then cancels
    * the subscription.

shared/src/main/scala/monifu/reactive/operators/switch.scala

@@ -16,64 +16,78 @@
 
 package monifu.reactive.operators
 
-import monifu.concurrent.cancelables.{BooleanCancelable, RefCountCancelable}
+import monifu.concurrent.cancelables.{RefCountCancelable, SerialCancelable}
 import monifu.reactive.Ack.{Cancel, Continue}
 import monifu.reactive._
 import monifu.reactive.internals._
 import monifu.reactive.observers.SynchronousObserver
+import scala.collection.mutable
 import scala.concurrent.Future
 
 object switch {
   /**
    * Implementation for [[Observable.concat]].
    */
-  def apply[T, U](source: Observable[T])(implicit ev: T <:< Observable[U]): Observable[U] =
+  def apply[T, U](source: Observable[T], delayErrors: Boolean)
+      (implicit ev: T <:< Observable[U]): Observable[U] = {
+
     Observable.create { subscriber:Subscriber[U] =>
       implicit val s = subscriber.scheduler
       val observerU = subscriber.observer
 
       source.unsafeSubscribe(new SynchronousObserver[T] { self =>
         // Global subscription, is canceled by the downstream
         // observer and if canceled all streaming is supposed to stop
-        private[this] val upstream = BooleanCancelable()
+        private[this] val upstream = SerialCancelable()
 
         // MUST BE synchronized by `self`
         private[this] var ack: Future[Ack] = Continue
-        // To be accessed only in `self.onNext`
-        private[this] var current: BooleanCancelable = null
+        // MUST BE synchronized by `self`
+        private[this] val errors = if (delayErrors)
+          mutable.ArrayBuffer.empty[Throwable] else null
 
         private[this] val refCount = RefCountCancelable {
           self.synchronized {
-            ack.onContinueSignalComplete(observerU)
+            if (delayErrors && errors.nonEmpty) {
+              ack.onContinueSignalError(observerU, CompositeException(errors))
+              errors.clear()
+            }
+            else {
+              ack.onContinueSignalComplete(observerU)
+            }
           }
         }
 
         def onNext(childObservable: T) = {
           if (upstream.isCanceled) Cancel else {
             // canceling current observable in order to
             // start the new stream
-            val activeSubscription = refCount.acquire()
-            if (current != null) current.cancel()
-            current = activeSubscription
+            val refID = refCount.acquire()
+            upstream := refID
 
             childObservable.unsafeSubscribe(new Observer[U] {
               def onNext(elem: U) = self.synchronized {
-                if (upstream.isCanceled || activeSubscription.isCanceled)
-                  Cancel
-                else {
+                if (refID.isCanceled) Cancel else {
                   ack = ack.onContinueStreamOnNext(observerU, elem)
                   ack.ifCanceledDoCancel(upstream)
-                  ack
                 }
               }
 
-              def onError(ex: Throwable): Unit =
-                self.onError(ex)
+              def onError(ex: Throwable): Unit = {
+                if (delayErrors) self.synchronized {
+                  errors += ex
+                  onComplete()
+                }
+                else {
+                  self.onError(ex)
+                }
+              }
 
               def onComplete(): Unit = {
-                // NOTE: we aren't sending this onComplete signal downstream to our observerU
-                // we will eventually do that after all of them are complete
-                activeSubscription.cancel()
+                // NOTE: we aren't sending this onComplete signal downstream to
+                // our observerU we will eventually do that after all of them
+                // are complete
+                refID.cancel()
               }
             })
 
@@ -83,7 +97,11 @@ object switch {
 
         def onError(ex: Throwable): Unit =
           self.synchronized {
-            if (!upstream.isCanceled) {
+            if (delayErrors) {
+              errors += ex
+              onComplete()
+            }
+            else if (!upstream.isCanceled) {
               upstream.cancel()
               ack.onContinueSignalError(observerU, ex)
             }
@@ -94,4 +112,5 @@ object switch {
         }
       })
     }
+  }
 }

shared/src/test/scala/monifu/concurrent/cancelables/SerialCancelableSuite.scala

@@ -0,0 +1,69 @@
+/*
+ * Copyright (c) 2014-2015 Alexandru Nedelcu
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package monifu.concurrent.cancelables
+
+import minitest.SimpleTestSuite
+
+object SerialCancelableSuite extends SimpleTestSuite {
+  test("cancel()") {
+    var effect = 0
+    val sub = BooleanCancelable(effect += 1)
+    val mSub = SerialCancelable(sub)
+
+    assert(effect == 0)
+    assert(!sub.isCanceled)
+    assert(!mSub.isCanceled)
+
+    mSub.cancel()
+    assert(sub.isCanceled && mSub.isCanceled)
+    assert(effect == 1)
+
+    mSub.cancel()
+    assert(sub.isCanceled && mSub.isCanceled)
+    assert(effect == 1)
+  }
+
+  test("cancel() after second assignment") {
+    var effect = 0
+    val sub = BooleanCancelable(effect += 1)
+    val mSub = SerialCancelable(sub)
+    val sub2 = BooleanCancelable(effect += 10)
+    mSub() = sub2
+
+    assert(effect == 1)
+    assert(sub.isCanceled && !sub2.isCanceled && !mSub.isCanceled)
+
+    mSub.cancel()
+    assert(sub2.isCanceled && mSub.isCanceled && sub.isCanceled)
+    assertEquals(effect, 11)
+  }
+
+  test("automatically cancel assigned") {
+    val mSub = SerialCancelable()
+    mSub.cancel()
+
+    var effect = 0
+    val sub = BooleanCancelable(effect += 1)
+
+    assert(effect == 0)
+    assert(!sub.isCanceled && mSub.isCanceled)
+
+    mSub() = sub
+    assert(effect == 1)
+    assert(sub.isCanceled)
+  }
+}