Add 'awaitSingleOr*' methods to the JDK9 Flow integration

Author: dkhalanskyjb

reactive/kotlinx-coroutines-jdk9/api/kotlinx-coroutines-jdk9.api

@@ -5,6 +5,9 @@ public final class kotlinx/coroutines/jdk9/AwaitKt {
 	public static final fun awaitFirstOrNull (Ljava/util/concurrent/Flow$Publisher;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun awaitLast (Ljava/util/concurrent/Flow$Publisher;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun awaitSingle (Ljava/util/concurrent/Flow$Publisher;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun awaitSingleOrDefault (Ljava/util/concurrent/Flow$Publisher;Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun awaitSingleOrElse (Ljava/util/concurrent/Flow$Publisher;Lkotlin/jvm/functions/Function0;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun awaitSingleOrNull (Ljava/util/concurrent/Flow$Publisher;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 }
 
 public final class kotlinx/coroutines/jdk9/PublishKt {

reactive/kotlinx-coroutines-jdk9/src/Await.kt

@@ -79,3 +79,44 @@ public suspend fun <T> Flow.Publisher<T>.awaitLast(): T =
  */
 public suspend fun <T> Flow.Publisher<T>.awaitSingle(): T =
         FlowAdapters.toPublisher(this).awaitSingle()
+
+/**
+ * Awaits the single value from the given observable, or returns the [default] value if none is emitted, without
+ * blocking the thread, and returns the resulting value, or, if this observable has produced an error, throws the
+ * corresponding exception.
+ *
+ * This suspending function is cancellable.
+ * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
+ * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
+ *
+ * @throws NoSuchElementException if the publisher does not emit any value
+ * @throws IllegalArgumentException if the publisher emits more than one value
+ */
+public suspend fun <T> Flow.Publisher<T>.awaitSingleOrDefault(default: T): T =
+    FlowAdapters.toPublisher(this).awaitSingleOrDefault(default)
+
+/**
+ * Awaits the single value from the given observable without blocking the thread and returns the resulting value, or, if
+ * this observable has produced an error, throws the corresponding exception. If more than one value or none were
+ * produced by the publisher, `null` is returned.
+ *
+ * This suspending function is cancellable.
+ * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
+ * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
+ */
+public suspend fun <T> Flow.Publisher<T>.awaitSingleOrNull(): T? =
+    FlowAdapters.toPublisher(this).awaitSingleOrNull()
+
+/**
+ * Awaits the single value from the given observable, or calls [defaultValue] to get a value if none is emitted, without
+ * blocking the thread, and returns the resulting value, or, if this observable has produced an error, throws the
+ * corresponding exception.
+ *
+ * This suspending function is cancellable.
+ * If the [Job] of the current coroutine is cancelled or completed while the suspending function is waiting, this
+ * function immediately cancels its [Flow.Subscription] and resumes with [CancellationException].
+ *
+ * @throws IllegalArgumentException if the publisher emits more than one value
+ */
+public suspend fun <T> Flow.Publisher<T>.awaitSingleOrElse(defaultValue: () -> T): T =
+    FlowAdapters.toPublisher(this).awaitSingleOrElse(defaultValue)

reactive/kotlinx-coroutines-jdk9/test/IntegrationTest.kt

@@ -49,6 +49,9 @@ class IntegrationTest(
         assertEquals("ELSE", pub.awaitFirstOrElse { "ELSE" })
         assertFailsWith<NoSuchElementException> { pub.awaitLast() }
         assertFailsWith<NoSuchElementException> { pub.awaitSingle() }
+        assertEquals("OK", pub.awaitSingleOrDefault("OK"))
+        assertNull(pub.awaitSingleOrNull())
+        assertEquals("ELSE", pub.awaitSingleOrElse { "ELSE" })
         var cnt = 0
         pub.collect { cnt++ }
         assertEquals(0, cnt)
@@ -66,6 +69,9 @@ class IntegrationTest(
         assertEquals("OK", pub.awaitFirstOrElse { "ELSE" })
         assertEquals("OK", pub.awaitLast())
         assertEquals("OK", pub.awaitSingle())
+        assertEquals("OK", pub.awaitSingleOrDefault("!"))
+        assertEquals("OK", pub.awaitSingleOrNull())
+        assertEquals("OK", pub.awaitSingleOrElse { "ELSE" })
         var cnt = 0
         pub.collect {
             assertEquals("OK", it)
@@ -85,10 +91,13 @@ class IntegrationTest(
         }
         assertEquals(1, pub.awaitFirst())
         assertEquals(1, pub.awaitFirstOrDefault(0))
-        assertEquals(n, pub.awaitLast())
         assertEquals(1, pub.awaitFirstOrNull())
         assertEquals(1, pub.awaitFirstOrElse { 0 })
+        assertEquals(null, pub.awaitSingleOrNull())
+        assertEquals(n, pub.awaitLast())
         assertFailsWith<IllegalArgumentException> { pub.awaitSingle() }
+        assertFailsWith<IllegalArgumentException> { pub.awaitSingleOrDefault(0) }
+        assertFailsWith<IllegalArgumentException> { pub.awaitSingleOrElse { 0 } }
         checkNumbers(n, pub)
         val flow = pub.asFlow()
         checkNumbers(n, flow.flowOn(ctx(coroutineContext)).asPublisher())
@@ -107,7 +116,7 @@ class IntegrationTest(
     }
 
     @Test
-    fun testEmptySingle() = runTest(unhandled = listOf { e -> e is NoSuchElementException}) {
+    fun testEmptySingle() = runTest(unhandled = listOf { e -> e is NoSuchElementException }) {
         expect(1)
         val job = launch(Job(), start = CoroutineStart.UNDISPATCHED) {
             flowPublish<String> {