Fixed CompletableSource.await and added test for it

Author: elizarov

kotlinx-coroutines-rx2/src/main/kotlin/kotlinx/coroutines/experimental/rx2/RxAwait.kt

@@ -36,7 +36,7 @@ public suspend fun CompletableSource.await(): Unit = suspendCancellableCoroutine
     subscribe(object : CompletableObserver {
         override fun onSubscribe(d: Disposable) { cont.disposeOnCompletion(d) }
         override fun onComplete() { cont.resume(Unit) }
-        override fun onError(e: Throwable) { cont.tryResumeWithException(e) }
+        override fun onError(e: Throwable) { cont.resumeWithException(e) }
     })
 }
 

kotlinx-coroutines-rx2/src/test/kotlin/kotlinx/coroutines/experimental/rx2/CompletableTest.kt

@@ -16,7 +16,6 @@
 
 package kotlinx.coroutines.experimental.rx2
 
-import kotlinx.coroutines.experimental.CancellationException
 import kotlinx.coroutines.experimental.TestBase
 import kotlinx.coroutines.experimental.runBlocking
 import kotlinx.coroutines.experimental.yield
@@ -83,4 +82,32 @@ class CompletableTest : TestBase() {
         yield()
         finish(6)
     }
+
+    @Test
+    fun testAwaitSuccess() = runBlocking<Unit> {
+        expect(1)
+        val completable = rxCompletable(context) {
+            expect(3)
+        }
+        expect(2)
+        completable.await() // 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.await() // shall launch coroutine and throw exception
+            expectUnreached()
+        } catch (e: RuntimeException) {
+            finish(4)
+            assertThat(e.message, IsEqual("OK"))
+        }
+    }
 }