Merge pull request #27 from Laimiux/lt/switchOnNext

Added extension method for static Observable.switchOnNext method

Author: MarioAriasC

src/main/kotlin/rx/lang/kotlin/observables.kt

@@ -106,3 +106,5 @@ public inline fun <T> Observable<T>.subscribeWith( body : FunctionSubscriberModi
     modifier.body()
     return subscribe(modifier.subscriber)
 }
+
+public fun <T> Observable<Observable<T>>.switchOnNext(): Observable<T> = Observable.switchOnNext(this)

src/test/kotlin/rx/lang/kotlin/ExtensionTests.kt

@@ -16,15 +16,15 @@
 
 package rx.lang.kotlin
 
-import rx.Observable
 import org.junit.Test
 import org.mockito.Mockito.*
 import org.mockito.Matchers.*
 import org.junit.Assert.*
-import rx.Notification
 import kotlin.concurrent.thread
-import rx.Subscriber
 import org.funktionale.partials.*
+import rx.*
+import rx.schedulers.TestScheduler
+import java.util.concurrent.TimeUnit
 
 /**
  * This class contains tests using the extension functions provided by the language adaptor.
@@ -236,6 +236,41 @@ public class ExtensionTests : KotlinTests() {
         assertEquals(listOf(3, 6, 9), values[2])
     }
 
+    @Test
+    public fun testSwitchOnNext() {
+        val testScheduler = TestScheduler()
+        val worker = testScheduler.createWorker()
+
+        val observable = observable<Observable<Long>> { s ->
+            fun at(delay: Long, func : () -> Unit){
+                worker.schedule({
+                    func()
+                }, delay, TimeUnit.MILLISECONDS)
+            }
+
+            val first = Observable.interval(5, TimeUnit.MILLISECONDS, testScheduler).take(3)
+            at(0, { s.onNext(first) })
+
+            val second = Observable.interval(5, TimeUnit.MILLISECONDS, testScheduler).take(3)
+            at(11, { s.onNext(second) })
+
+            at(40, { s.onCompleted() })
+        }
+
+        observable.switchOnNext().subscribe(received)
+
+        val inOrder = inOrder(a)
+        testScheduler.advanceTimeTo(10, TimeUnit.MILLISECONDS)
+        inOrder.verify(a, times(1)).received(0L)
+        inOrder.verify(a, times(1)).received(1L)
+
+        testScheduler.advanceTimeTo(40, TimeUnit.MILLISECONDS)
+        inOrder.verify(a, times(1)).received(0L)
+        inOrder.verify(a, times(1)).received(1L)
+        inOrder.verify(a, times(1)).received(2L)
+        inOrder.verifyNoMoreInteractions()
+    }
+
     val funOnSubscribe: (Int, Subscriber<in String>) -> Unit = { counter, subscriber ->
         subscriber.onNext("hello_$counter")
         subscriber.onCompleted()