Introduce Flow.onEmpty operator

Fixes #1890

Author: qwwdfsad

kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

@@ -942,6 +942,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final synthetic fun onCompletion (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun onCompletion (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun onEach (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun onEmpty (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun onErrorCollect (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
 	public static synthetic fun onErrorCollect$default (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun onErrorResume (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;

kotlinx-coroutines-core/common/src/flow/operators/Emitters.kt

@@ -155,6 +155,39 @@ public fun <T> Flow<T>.onCompletion(
     exception?.let { throw it }
 }
 
+/**
+ * Invokes the given [action] when this flow completes without emitting any elements.
+ * The receiver of the [action] is [FlowCollector], so `onEmpty` can emit additional elements.
+ *
+ * For example:
+ *
+ * ```
+ * emptyFlow<Int>().onEmpty {
+ *     emit(1)
+ *     emit(2)
+ * }
+ * .collect { println(it) } // prints 1, 2
+ * ```
+ */
+@ExperimentalCoroutinesApi
+public fun <T> Flow<T>.onEmpty(
+    action: suspend FlowCollector<T>.() -> Unit
+): Flow<T> = unsafeFlow {
+    var isEmpty = true
+    collect {
+        isEmpty = false
+        emit(it)
+    }
+    if (isEmpty) {
+        val collector = SafeCollector(this, coroutineContext)
+        try {
+            collector.action()
+        } finally {
+            collector.releaseIntercepted()
+        }
+    }
+}
+
 private class ThrowingCollector(private val e: Throwable) : FlowCollector<Any?> {
     override suspend fun emit(value: Any?) {
         throw e

kotlinx-coroutines-core/common/test/flow/operators/OnEmptyTest.kt

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.flow.operators
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import kotlin.test.*
+
+class OnEmptyTest : TestBase() {
+
+    @Test
+    fun testOnEmptyInvoked() = runTest {
+        val flow = emptyFlow<Int>().onEmpty { emit(1) }
+        assertEquals(1, flow.single())
+    }
+
+    @Test
+    fun testOnEmptyNotInvoked() = runTest {
+        val flow = flowOf(1).onEmpty { emit(2) }
+        assertEquals(1, flow.single())
+    }
+
+    @Test
+    fun testOnEmptyNotInvokedOnError() = runTest {
+        val flow = flow<Int> {
+            throw TestException()
+        }.onEmpty { expectUnreached() }
+        assertFailsWith<TestException>(flow)
+    }
+
+    @Test
+    fun testOnEmptyNotInvokedOnCancellation() = runTest {
+        val flow = flow<Int> {
+            expect(2)
+            hang { expect(4) }
+        }.onEmpty { expectUnreached() }
+
+        expect(1)
+        val job = flow.onEach { expectUnreached() }.launchIn(this)
+        yield()
+        expect(3)
+        job.cancelAndJoin()
+        finish(5)
+    }
+
+    @Test
+    fun testOnEmptyCancellation() = runTest {
+        val flow = emptyFlow<Int>().onEmpty {
+            expect(2)
+            hang { expect(4) }
+            emit(1)
+        }
+        expect(1)
+        val job = flow.onEach { expectUnreached() }.launchIn(this)
+        yield()
+        expect(3)
+        job.cancelAndJoin()
+        finish(5)
+    }
+
+    @Test
+    fun testOnEmptyTransparencyViolation() = runTest {
+        val flow = emptyFlow<Int>().onEmpty {
+            expect(2)
+            coroutineScope {
+                launch {
+                    try {
+                        emit(1)
+                    } catch (e: IllegalStateException) {
+                        expect(3)
+                    }
+                }
+            }
+        }
+        expect(1)
+        assertNull(flow.singleOrNull())
+        finish(4)
+    }
+}