Flow improvements:

  * Declaration-site variance for FlowCollector
  * Better Flow documentation
  * Use SendChannel.isEmpty in FlatMap, add nulls test, use unbox, fail fast for conflated channels
  * Simplify Flow.asFlowable in rx2 module
  * Consistent naming

Author: qwwdfsad

kotlinx-coroutines-core/common/src/flow/Builders.kt

@@ -44,9 +44,9 @@ import kotlin.jvm.*
  * If you want to switch the context where this flow is executed use [flowOn] operator.
  */
 @FlowPreview
-public fun <T> flow(@BuilderInference block: suspend FlowCollector<in T>.() -> Unit): Flow<T> {
+public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> {
     return object : Flow<T> {
-        override suspend fun collect(collector: FlowCollector<in T>) {
+        override suspend fun collect(collector: FlowCollector<T>) {
             SafeCollector(collector, coroutineContext).block()
         }
     }
@@ -58,9 +58,9 @@ public fun <T> flow(@BuilderInference block: suspend FlowCollector<in T>.() -> U
  */
 @FlowPreview
 @PublishedApi
-internal fun <T> unsafeFlow(@BuilderInference block: suspend FlowCollector<in T>.() -> Unit): Flow<T> {
+internal fun <T> unsafeFlow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> {
     return object : Flow<T> {
-        override suspend fun collect(collector: FlowCollector<in T>) {
+        override suspend fun collect(collector: FlowCollector<T>) {
             collector.block()
         }
     }
@@ -129,7 +129,7 @@ public fun <T> flowOf(vararg elements: T): Flow<T> = unsafeFlow {
 public fun <T> emptyFlow(): Flow<T> = EmptyFlow
 
 private object EmptyFlow : Flow<Nothing> {
-    override suspend fun collect(collector: FlowCollector<in Nothing>) = Unit
+    override suspend fun collect(collector: FlowCollector<Nothing>) = Unit
 }
 
 /**

kotlinx-coroutines-core/common/src/flow/Flow.kt

@@ -88,16 +88,22 @@ public interface Flow<out T> {
      * 1) It should not change the coroutine context (e.g. with `withContext(Dispatchers.IO)`) when emitting values.
      *    The emission should happen in the context of the [collect] call.
      *
-     * Only coroutine builders that inherit the context are allowed, for example the following code is fine:
+     * Only coroutine builders that inherit the context are allowed, for example:
      * ```
-     * coroutineScope { // Context is inherited
-     *     launch { // Dispatcher is not overridden, fine as well
-     *         collector.emit(someValue)
+     * class MyFlow : Flow<Int> {
+     *     override suspend fun collect(collector: FlowCollector<Int>) {
+     *         coroutineScope {
+     *             // Context is inherited
+     *             launch { // Dispatcher is not overridden, fine as well
+     *                 collector.emit(42) // Emit from the launched coroutine
+     *             }
+     *         }
      *     }
      * }
      * ```
+     * is a proper [Flow] implementation, but using `launch(Dispatchers.IO)` is not.
      *
      * 2) It should serialize calls to [emit][FlowCollector.emit] as [FlowCollector] implementations are not thread safe by default.
      */
-    public suspend fun collect(collector: FlowCollector<in T>)
+    public suspend fun collect(collector: FlowCollector<T>)
 }

kotlinx-coroutines-core/common/src/flow/FlowCollector.kt

@@ -14,7 +14,7 @@ import kotlinx.coroutines.*
  * Implementations of this interface are not thread-safe.
  */
 @FlowPreview
-public interface FlowCollector<T> {
+public interface FlowCollector<in T> {
 
     /**
      * Collects the value emitted by the upstream.

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

@@ -29,7 +29,7 @@ import kotlinx.coroutines.flow.unsafeFlow as flow
  *         .single() // Will be executed in the Main
  * }
  * ```
- * For more explanation of purity concept please refer to [Flow] documentation.
+ * For more explanation of context preservation please refer to [Flow] documentation.
  *
  * This operator uses a channel of the specific [bufferSize] in order to switch between contexts,
  * but it is not guaranteed that the channel will be created, implementation is free to optimize it away in case of fusing.
@@ -83,7 +83,7 @@ public fun <T> Flow<T>.flowOn(flowContext: CoroutineContext, bufferSize: Int = 1
  *     }
  *     .map { ... } // Not affected
  * ```
- * For more explanation of purity concept please refer to [Flow] documentation.
+ * For more explanation of context preservation please refer to [Flow] documentation.
  *
  * This operator uses channel of the specific [bufferSize] in order to switch between contexts,
  * but it is not guaranteed that channel will be created, implementation is free to optimize it away in case of fusing.

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

@@ -74,7 +74,7 @@ public fun <T> Flow<T>.retry(
     }
 }
 
-private fun <T> Flow<T>.collectSafely(onException: suspend FlowCollector<in T>.(Throwable) -> Unit): Flow<T> =
+private fun <T> Flow<T>.collectSafely(onException: suspend FlowCollector<T>.(Throwable) -> Unit): Flow<T> =
     flow {
         // Note that exception may come from the downstream operators, we should not switch on that
         var fromDownstream = false

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

@@ -16,23 +16,23 @@ import kotlin.jvm.*
 import kotlinx.coroutines.flow.unsafeFlow as flow
 
 /**
- * Transforms elements emitted by the original flow by applying [mapper], that returns another flow, and then concatenating and flattening these flows.
+ * Transforms elements emitted by the original flow by applying [transform], that returns another flow, and then concatenating and flattening these flows.
  * This method is identical to `flatMapMerge(concurrency = 1, bufferSize = 1)`
  *
  * Note that even though this operator looks very familiar, we discourage its usage in a regular application-specific flows.
  * Most likely, suspending operation in [map] operator will be sufficient and linear transformations are much easier to reason about.
  */
 @FlowPreview
-public fun <T, R> Flow<T>.flatMapConcat(mapper: suspend (value: T) -> Flow<R>): Flow<R> = flow {
+public fun <T, R> Flow<T>.flatMapConcat(transform: suspend (value: T) -> Flow<R>): Flow<R> = flow {
     collect { value ->
-        mapper(value).collect { innerValue ->
+        transform(value).collect { innerValue ->
             emit(innerValue)
         }
     }
 }
 
 /**
- * Transforms elements emitted by the original flow by applying [mapper], that returns another flow, and then merging and flattening these flows.
+ * Transforms elements emitted by the original flow by applying [transform], that returns another flow, and then merging and flattening these flows.
  *
  * Note that even though this operator looks very familiar, we discourage its usage in a regular application-specific flows.
  * Most likely, suspending operation in [map] operator will be sufficient and linear transformations are much easier to reason about.
@@ -41,14 +41,17 @@ public fun <T, R> Flow<T>.flatMapConcat(mapper: suspend (value: T) -> Flow<R>):
  * [concurrency] parameter controls the size of in-flight flows, at most [concurrency] flows are collected at the same time.
  */
 @FlowPreview
-public fun <T, R> Flow<T>.flatMapMerge(concurrency: Int = 16, bufferSize: Int = 16, mapper: suspend (value: T) -> Flow<R>): Flow<R> {
+public fun <T, R> Flow<T>.flatMapMerge(concurrency: Int = 16, bufferSize: Int = 16, transform: suspend (value: T) -> Flow<R>): Flow<R> {
+    require(bufferSize >= 0) { "Expected non-negative buffer size, but had $bufferSize" }
+    require(concurrency >= 0) { "Expected non-negative concurrency level, but had $concurrency" }
     return flow {
         val semaphore = Channel<Unit>(concurrency)
         val flatMap = SerializingFlatMapCollector(this, bufferSize)
         coroutineScope {
             collect { outerValue ->
+                // TODO real semaphore (#94)
                 semaphore.send(Unit) // Acquire concurrency permit
-                val inner = mapper(outerValue)
+                val inner = transform(outerValue)
                 launch {
                     try {
                         inner.collect { value ->
@@ -94,13 +97,12 @@ private class SerializingFlatMapCollector<T>(
 ) {
 
     // Let's try to leverage the fact that flatMapMerge is never contended
-    private val channel: Channel<Any?> by lazy { Channel<Any?>(bufferSize) } // Should be any, but KT-30796
+    // TODO 1.2.1 do not allocate channel
+    private val channel = Channel<Any?>(bufferSize) // Should be any, but KT-30796
     private val inProgressLock = atomic(false)
-    private val sentValues = atomic(0)
 
     public suspend fun emit(value: T) {
         if (!inProgressLock.tryAcquire()) {
-            sentValues.incrementAndGet()
             channel.send(value ?: NullSurrogate)
             if (inProgressLock.tryAcquire()) {
                 helpEmit()
@@ -116,17 +118,14 @@ private class SerializingFlatMapCollector<T>(
     private suspend fun helpEmit() {
         while (true) {
             var element = channel.poll()
-            while (element != null) { // TODO receive or closed
-                if (element === NullSurrogate) downstream.emit(null as T)
-                else downstream.emit(element as T)
-                sentValues.decrementAndGet()
+            while (element != null) { // TODO receive or closed (#330)
+                downstream.emit(NullSurrogate.unbox(element))
                 element = channel.poll()
             }
 
             inProgressLock.release()
-            // Enforce liveness of the algorithm
-            // TODO looks like isEmpty use-case
-            if (sentValues.value == 0 || !inProgressLock.tryAcquire()) break
+            // Enforce liveness
+            if (channel.isEmpty || !inProgressLock.tryAcquire()) break
         }
     }
 }

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

@@ -26,7 +26,7 @@ import kotlinx.coroutines.flow.unsafeFlow as flow
  * ```
  */
 @FlowPreview
-public fun <T, R> Flow<T>.transform(@BuilderInference transformer: suspend FlowCollector<in R>.(value: T) -> Unit): Flow<R> {
+public fun <T, R> Flow<T>.transform(@BuilderInference transformer: suspend FlowCollector<R>.(value: T) -> Unit): Flow<R> {
     return flow {
         collect { value ->
             transformer(value)

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

@@ -7,50 +7,53 @@ package kotlinx.coroutines.flow
 import kotlinx.coroutines.*
 import kotlin.test.*
 
+/*
+ * Replace:  { i, j -> i + j } ->  { i, j -> i + j } as soon as KT-30991 is fixed
+ */
 class CombineLatestTest : TestBase() {
 
     @Test
     fun testCombineLatest() = runTest {
         val flow = flowOf("a", "b", "c")
         val flow2 = flowOf(1, 2, 3)
-        val list = flow.combineLatest(flow2, ::sum).toList()
+        val list = flow.combineLatest(flow2,  { i, j -> i + j }).toList()
         assertEquals(listOf("a1", "b1", "b2", "c2", "c3"), list)
     }
 
     @Test
     fun testNulls() = runTest {
         val flow = flowOf("a", null, null)
         val flow2 = flowOf(1, 2, 3)
-        val list = flow.combineLatest(flow2, ::sum).toList()
+        val list = flow.combineLatest(flow2,  { i, j -> i + j }).toList()
         assertEquals(listOf("a1", "null1", "null2", "null2", "null3"), list)
     }
 
     @Test
     fun testNullsOther() = runTest {
         val flow = flowOf("a", "b", "c")
         val flow2 = flowOf(null, 2, null)
-        val list = flow.combineLatest(flow2, ::sum).toList()
+        val list = flow.combineLatest(flow2,  { i, j -> i + j }).toList()
         assertEquals(listOf("anull", "bnull", "b2", "c2", "cnull"), list)
     }
 
     @Test
     fun testEmptyFlow() = runTest {
-        val flow = emptyFlow<String>().combineLatest(emptyFlow(), ::sum)
+        val flow = emptyFlow<String>().combineLatest(emptyFlow<Int>(),  { i, j -> i + j })
         assertNull(flow.singleOrNull())
     }
 
     @Test
     fun testFirstIsEmpty() = runTest {
         val f1 = emptyFlow<String>()
         val f2 = flowOf(1)
-        assertEquals(emptyList(), f1.combineLatest(f2, ::sum).toList())
+        assertEquals(emptyList(), f1.combineLatest(f2,  { i, j -> i + j }).toList())
     }
 
     @Test
     fun testSecondIsEmpty() = runTest {
         val f1 = flowOf("a")
         val f2 = emptyFlow<Int>()
-        assertEquals(emptyList(), f1.combineLatest(f2, ::sum).toList())
+        assertEquals(emptyList(), f1.combineLatest(f2,  { i, j -> i + j }).toList())
     }
 
     @Test
@@ -77,7 +80,7 @@ class CombineLatestTest : TestBase() {
             emit(3)
         }
 
-        val result = f1.combineLatest(f2, ::sum).toList()
+        val result = f1.combineLatest(f2,  { i, j -> i + j }).toList()
         assertEquals(listOf("a1", "b1", "c1", "c2", "c3"), result)
         finish(8)
     }

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

@@ -2,7 +2,7 @@
  * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
-package kotlinx.coroutines.flow.operators
+package kotlinx.coroutines.flow
 
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
@@ -44,6 +44,15 @@ abstract class FlatMapBaseTest : TestBase() {
         assertEquals(42, value)
     }
 
+    @Test
+    fun testNulls() = runTest {
+        val list = flowOf(1, null, 2).flatMap {
+            flowOf(1, null, null, 2)
+        }.toList()
+
+        assertEquals(List(3) { listOf(1, null, null, 2)}.flatten(), list)
+    }
+
     @Test
     fun testContext() = runTest {
         val captured = ArrayList<String>()

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

@@ -5,13 +5,11 @@
 package kotlinx.coroutines.flow
 
 import kotlinx.coroutines.*
-import kotlinx.coroutines.channels.*
-import kotlinx.coroutines.flow.operators.*
 import kotlin.test.*
 
 class FlatMapConcatTest : FlatMapBaseTest() {
 
-    override fun <T> Flow<T>.flatMap(mapper: suspend (T) -> Flow<T>): Flow<T> = flatMapConcat(mapper = mapper)
+    override fun <T> Flow<T>.flatMap(mapper: suspend (T) -> Flow<T>): Flow<T> = flatMapConcat(transform = mapper)
 
     @Test
     fun testFlatMapConcurrency() = runTest {

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

@@ -3,11 +3,10 @@
  */
 
 
-package kotlinx.coroutines.flow.operators
+package kotlinx.coroutines.flow
 
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
-import kotlinx.coroutines.flow.*
 import kotlin.test.*
 
 abstract class FlatMapMergeBaseTest : FlatMapBaseTest() {

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

@@ -5,13 +5,11 @@
 package kotlinx.coroutines.flow
 
 import kotlinx.coroutines.*
-import kotlinx.coroutines.channels.*
-import kotlinx.coroutines.flow.operators.*
 import kotlin.test.*
 
 class FlatMapMergeTest : FlatMapMergeBaseTest() {
 
-    override fun <T> Flow<T>.flatMap(mapper: suspend (T) -> Flow<T>): Flow<T> = flatMapMerge(mapper = mapper)
+    override fun <T> Flow<T>.flatMap(mapper: suspend (T) -> Flow<T>): Flow<T> = flatMapMerge(transform = mapper)
 
     @Test
     override fun testFlatMapConcurrency() = runTest {

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

@@ -5,7 +5,6 @@
 package kotlinx.coroutines.flow
 
 import kotlinx.coroutines.*
-import kotlinx.coroutines.flow.operators.*
 import kotlin.test.*
 
 class FlattenConcatTest : FlatMapBaseTest() {

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

@@ -5,7 +5,6 @@
 package kotlinx.coroutines.flow
 
 import kotlinx.coroutines.*
-import kotlinx.coroutines.flow.operators.*
 import kotlin.test.*
 
 class FlattenMergeTest : FlatMapMergeBaseTest() {