#49 Rx1 Completable.awaitCompleted implementation. (#50)

#49 Rx1 Completable.awaitCompleted implementation.

Author: konrad-kaminski

reactive/kotlinx-coroutines-rx1/README.md

@@ -14,6 +14,7 @@ Suspending extension functions and suspending iteration:
 
 | **Name** | **Description**
 | -------- | ---------------
+| [Completable.awaitCompleted][rx.Completable.awaitCompleted] | Awaits for completion of the completable value 
 | [Single.await][rx.Single.await] | Awaits for completion of the single value and returns it 
 | [Observable.awaitFirst][rx.Observable.awaitFirst] | Returns the first value from the given observable
 | [Observable.awaitFirstOrDefault][rx.Observable.awaitFirstOrDefault] | Returns the first value from the given observable or default
@@ -44,6 +45,7 @@ Conversion functions:
 [rxCompletable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx1/kotlinx.coroutines.experimental.rx1/rx-completable.html
 [rxSingle]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx1/kotlinx.coroutines.experimental.rx1/rx-single.html
 [rxObservable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx1/kotlinx.coroutines.experimental.rx1/rx-observable.html
+[rx.Completable.awaitCompleted]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx1/kotlinx.coroutines.experimental.rx1/rx.-completable/await-completed.html
 [rx.Single.await]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx1/kotlinx.coroutines.experimental.rx1/rx.-single/await.html
 [rx.Observable.awaitFirst]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx1/kotlinx.coroutines.experimental.rx1/rx.-observable/await-first.html
 [rx.Observable.awaitFirstOrDefault]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx1/kotlinx.coroutines.experimental.rx1/rx.-observable/await-first-or-default.html

reactive/kotlinx-coroutines-rx1/src/main/kotlin/kotlinx/coroutines/experimental/rx1/RxAwait.kt

@@ -22,6 +22,23 @@ import kotlinx.coroutines.experimental.Job
 import kotlinx.coroutines.experimental.suspendCancellableCoroutine
 import rx.*
 
+// ------------------------ Completable ------------------------
+
+/**
+ * Awaits for completion of this completable without blocking a thread.
+ * Returns `Unit` or throws the corresponding exception if this completable had produced error.
+ *
+ * This suspending function is cancellable. If the [Job] of the invoking coroutine is completed while this
+ * suspending function is suspended, this function immediately resumes with [CancellationException].
+ */
+public suspend fun Completable.awaitCompleted(): Unit = suspendCancellableCoroutine { cont ->
+    subscribe(object : CompletableSubscriber {
+        override fun onSubscribe(s: Subscription) { cont.unsubscribeOnCompletion(s) }
+        override fun onCompleted() { cont.resume(Unit) }
+        override fun onError(e: Throwable) { cont.resumeWithException(e) }
+    })
+}
+
 // ------------------------ Single ------------------------
 
 /**

reactive/kotlinx-coroutines-rx1/src/test/kotlin/kotlinx/coroutines/experimental/rx1/CompletableTest.kt

@@ -83,4 +83,32 @@ class CompletableTest : TestBase() {
         yield()
         finish(7)
     }
+
+    @Test
+    fun testAwaitSuccess() = runBlocking<Unit> {
+        expect(1)
+        val completable = rxCompletable(context) {
+            expect(3)
+        }
+        expect(2)
+        completable.awaitCompleted() // shall launch coroutine
+        finish(4)
+    }
+
+    @Test
+    fun testAwaitFailure() = runBlocking<Unit> {
+        expect(1)
+        val completable = rxCompletable(context) {
+            expect(3)
+            throw RuntimeException("OK")
+        }
+        expect(2)
+        try {
+            completable.awaitCompleted() // shall launch coroutine and throw exception
+            expectUnreached()
+        } catch (e: RuntimeException) {
+            finish(4)
+            assertThat(e.message, IsEqual("OK"))
+        }
+    }
 }