Merge pull request #39 from cypressious/rx-firstOrDefault

Implement awaitFirstOrDefault for RxJava1

Author: elizarov

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

@@ -51,6 +51,16 @@ public suspend fun <T> Single<T>.await(): T = suspendCancellableCoroutine { cont
  */
 public suspend fun <T> Observable<T>.awaitFirst(): T = first().awaitOne()
 
+/**
+ * Awaits for the first value from the given observable or the [default] value if none is emitted without blocking a
+ * thread and returns the resulting value or throws the corresponding exception if this observable had produced error.
+ *
+ * This suspending function is cancellable.
+ * If the [Job] of the current coroutine is completed while this suspending function is waiting, this function
+ * immediately resumes with [CancellationException].
+ */
+public suspend fun <T> Observable<T>.awaitFirstOrDefault(default: T): T = firstOrDefault(default).awaitOne()
+
 /**
  * Awaits for the last value from the given observable without blocking a thread and
  * returns the resulting value or throws the corresponding exception if this observable had produced error.

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

@@ -90,6 +90,18 @@ class ObservableSingleTest {
 
     @Test
     fun testAwaitFirst() {
+        val observable = rxObservable(CommonPool) {
+            send(Observable.empty<String>().awaitFirstOrDefault("O") + "K")
+        }
+
+        checkSingleValue(observable) {
+            assertEquals("OK", it)
+        }
+    }
+
+
+    @Test
+    fun testAwaitFirstOrDefault() {
         val observable = rxObservable(CommonPool) {
             send(Observable.just("O", "#").awaitFirst() + "K")
         }