rxFlowable builder for Rx 2.x

Author: elizarov

reactive/kotlinx-coroutines-rx2/README.md

@@ -9,6 +9,7 @@ Coroutine builders:
 | [rxCompletable] | `Completable`                           | [CoroutineScope] | Cold completable that starts coroutine on subscribe
 | [rxSingle]      | `Single`                                | [CoroutineScope] | Cold single that starts coroutine on subscribe
 | [rxObservable]  | `Observable`                            | [ProducerScope]  | Cold observable that starts coroutine on subscribe
+| [rxFlowable]    | `Flowable`                              | [ProducerScope]  | Cold observable that starts coroutine on subscribe with **backpressure** support 
 
 Suspending extension functions and suspending iteration:
 
@@ -50,6 +51,7 @@ Conversion functions:
 [rxCompletable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/rx-completable.html
 [rxSingle]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/rx-single.html
 [rxObservable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/rx-observable.html
+[rxFlowable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/rx-flowable.html
 [io.reactivex.CompletableSource.await]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/io.reactivex.-completable-source/await.html
 [io.reactivex.SingleSource.await]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/io.reactivex.-single-source/await.html
 [io.reactivex.ObservableSource.awaitFirst]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.experimental.rx2/io.reactivex.-observable-source/await-first.html

reactive/kotlinx-coroutines-rx2/src/main/kotlin/kotlinx/coroutines/experimental/rx2/RxFlowable.kt

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * 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 kotlinx.coroutines.experimental.rx2
+
+import io.reactivex.Flowable
+import kotlinx.coroutines.experimental.channels.ProducerScope
+import kotlinx.coroutines.experimental.reactive.publish
+import kotlin.coroutines.experimental.CoroutineContext
+
+/**
+ * Creates cold [flowable][Flowable] that will run a given [block] in a coroutine.
+ * Every time the returned flowable is subscribed, it starts a new coroutine in the specified [context].
+ * Coroutine emits items with `send`. Unsubscribing cancels running coroutine.
+ *
+ * Invocations of `send` are suspended appropriately when subscribers apply back-pressure and to ensure that
+ * `onNext` is not invoked concurrently.
+ *
+ * | **Coroutine action**                         | **Signal to subscriber**
+ * | -------------------------------------------- | ------------------------
+ * | `send`                                       | `onNext`
+ * | Normal completion or `close` without cause   | `onComplete`
+ * | Failure with exception or `close` with cause | `onError`
+ */
+public fun <T> rxFlowable(
+    context: CoroutineContext,
+    block: suspend ProducerScope<T>.() -> Unit
+): Flowable<T> = Flowable.fromPublisher(publish(context, block))

reactive/kotlinx-coroutines-rx2/src/main/kotlin/kotlinx/coroutines/experimental/rx2/RxObservable.kt

@@ -38,8 +38,8 @@ import kotlin.coroutines.experimental.startCoroutine
  * Every time the returned observable is subscribed, it starts a new coroutine in the specified [context].
  * Coroutine emits items with `send`. Unsubscribing cancels running coroutine.
  *
- * Invocations of `send` are suspended appropriately when subscribers apply back-pressure and to ensure that
- * `onNext` is not invoked concurrently.
+ * Invocations of `send` are suspended appropriately to ensure that `onNext` is not invoked concurrently.
+ * Note, that Rx 2.x [Observable] **does not support backpressure**. Use [rxFlowable].
  *
  * | **Coroutine action**                         | **Signal to subscriber**
  * | -------------------------------------------- | ------------------------

reactive/kotlinx-coroutines-rx2/src/test/kotlin/kotlinx/coroutines/experimental/rx2/FlowableTest.kt

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2016-2017 JetBrains s.r.o.
+ *
+ * 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 kotlinx.coroutines.experimental.rx2
+
+import kotlinx.coroutines.experimental.TestBase
+import kotlinx.coroutines.experimental.runBlocking
+import kotlinx.coroutines.experimental.yield
+import org.hamcrest.core.IsEqual
+import org.hamcrest.core.IsInstanceOf
+import org.junit.Assert
+import org.junit.Test
+
+class FlowableTest : TestBase() {
+    @Test
+    fun testBasicSuccess() = runBlocking<Unit> {
+        expect(1)
+        val observable = rxFlowable(context) {
+            expect(4)
+            send("OK")
+        }
+        expect(2)
+        observable.subscribe { value ->
+            expect(5)
+            Assert.assertThat(value, IsEqual("OK"))
+        }
+        expect(3)
+        yield() // to started coroutine
+        finish(6)
+    }
+
+    @Test
+    fun testBasicFailure() = runBlocking<Unit> {
+        expect(1)
+        val observable = rxFlowable<String>(context) {
+            expect(4)
+            throw RuntimeException("OK")
+        }
+        expect(2)
+        observable.subscribe({
+            expectUnreached()
+        }, { error ->
+            expect(5)
+            Assert.assertThat(error, IsInstanceOf(RuntimeException::class.java))
+            Assert.assertThat(error.message, IsEqual("OK"))
+        })
+        expect(3)
+        yield() // to started coroutine
+        finish(6)
+    }
+
+    @Test
+    fun testBasicUnsubscribe() = runBlocking<Unit> {
+        expect(1)
+        val observable = rxFlowable<String>(context) {
+            expect(4)
+            yield() // back to main, will get cancelled
+            expectUnreached()
+        }
+        expect(2)
+        val sub = observable.subscribe({
+            expectUnreached()
+        }, {
+            expectUnreached()
+        })
+        expect(3)
+        yield() // to started coroutine
+        expect(5)
+        sub.dispose() // will cancel coroutine
+        yield()
+        finish(6)
+    }
+}