Issue monix#42 - switch operator WIP

Author: aoprisan

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

@@ -31,14 +31,14 @@ final class RefCountCancelable private (onCancel: () => Unit) extends BooleanCan
     state.get.isCanceled
 
   @tailrec
-  def acquire(): Cancelable = {
+  def acquire(): BooleanCancelable = {
     val oldState = state.get
     if (oldState.isCanceled)
       BooleanCancelable.alreadyCanceled
     else if (!state.compareAndSet(oldState, oldState.copy(activeCounter = oldState.activeCounter + 1)))
       acquire()
     else
-      Cancelable {
+      BooleanCancelable {
         val newState = state.transformAndGet(s => s.copy(activeCounter = s.activeCounter - 1))
         if (newState.activeCounter == 0 && newState.isCanceled)
           onCancel()

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

@@ -1168,6 +1168,13 @@ 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

@@ -0,0 +1,97 @@
+/*
+ * 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.reactive.operators
+
+import monifu.concurrent.cancelables.{BooleanCancelable, RefCountCancelable}
+import monifu.reactive.Ack.{Cancel, Continue}
+import monifu.reactive._
+import monifu.reactive.internals._
+import monifu.reactive.observers.SynchronousObserver
+import scala.concurrent.Future
+
+object switch {
+  /**
+   * Implementation for [[Observable.concat]].
+   */
+  def apply[T, U](source: Observable[T])(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()
+
+        // 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
+
+        private[this] val refCount = RefCountCancelable {
+          self.synchronized {
+            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
+
+            childObservable.unsafeSubscribe(new Observer[U] {
+              def onNext(elem: U) = self.synchronized {
+                if (upstream.isCanceled || activeSubscription.isCanceled)
+                  Cancel
+                else {
+                  ack = ack.onContinueStreamOnNext(observerU, elem)
+                  ack.ifCanceledDoCancel(upstream)
+                  ack
+                }
+              }
+
+              def onError(ex: Throwable): Unit =
+                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()
+              }
+            })
+
+            Continue
+          }
+        }
+
+        def onError(ex: Throwable): Unit =
+          self.synchronized {
+            if (!upstream.isCanceled) {
+              upstream.cancel()
+              ack.onContinueSignalError(observerU, ex)
+            }
+          }
+
+        def onComplete(): Unit = {
+          refCount.cancel()
+        }
+      })
+    }
+}

shared/src/test/scala/monifu/reactive/operators/MergeManySuite.scala

@@ -31,7 +31,7 @@ object MergeManySuite extends BaseOperatorSuite {
 
   def observableInError(sourceCount: Int, ex: Throwable) = Some {
     val o = createObservableEndingInError(Observable.range(0, sourceCount), ex)
-      .flatMap(i => Observable.fromIterable(Seq(i, i, i, i)))
+      .mergeMap(i => Observable.fromIterable(Seq(i, i, i, i)))
     Sample(o, count(sourceCount), sum(sourceCount), Zero, Zero)
   }
 
@@ -40,7 +40,7 @@ object MergeManySuite extends BaseOperatorSuite {
   }
 
   def brokenUserCodeObservable(sourceCount: Int, ex: Throwable) = Some {
-    val o = Observable.range(0, sourceCount).flatMap { i =>
+    val o = Observable.range(0, sourceCount).mergeMap { i =>
       if (i == sourceCount-1)
         throw ex
       else

shared/src/test/scala/monifu/reactive/operators/SwitchSuite.scala

@@ -0,0 +1,68 @@
+/*
+ * 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.reactive.operators
+
+import monifu.reactive.{Ack, Observer, Observable}
+import scala.concurrent.Future
+import scala.concurrent.duration._
+import scala.language.postfixOps
+
+object SwitchSuite extends BaseOperatorSuite {
+  def createChild() = {
+    Observable.interval(1.second).take(2) ++
+      Observable.interval(1.second).drop(3)
+  }
+
+  def observable(sourceCount: Int) = Some {
+    val o = Observable.interval(2.seconds)
+      .take(sourceCount)
+      .map(i => (if (i < sourceCount-1) createChild() else Observable.interval(1.second)).take(sourceCount))
+      .switch
+
+    val count = (sourceCount - 1) * 2 + sourceCount
+    val sum = (sourceCount - 1) + (1 until sourceCount).sum
+    Sample(o, count, sum, waitFirst, waitNext)
+  }
+
+  def waitFirst = 0.seconds
+  def waitNext = 1.second
+
+  def observableInError(sourceCount: Int, ex: Throwable) = Some {
+    def createChild(i: Long) = {
+      if (i < sourceCount-1)
+        (Observable.interval(1.second).take(2) ++
+          Observable.interval(1.second).drop(3)).doOnCanceled("CANCELED")
+      else
+        Observable.interval(1.second)
+          .take(sourceCount)
+    }
+
+    val source = Observable.interval(2.seconds).take(sourceCount)
+    val endingInError = createObservableEndingInError(source, ex)
+    val o = endingInError
+      .map(i => createChild(i))
+      .switch
+
+    val count = (sourceCount - 1) * 2
+    val sum = sourceCount - 1
+
+    Sample(o, count, sum, waitFirst, waitNext)
+  }
+
+  def brokenUserCodeObservable(sourceCount: Int, ex: Throwable) =
+    None
+}