Use regular produce instead of flowProduce in channelFlow

Concurrent work is already properly decomposed and does not expose an "partial cancellation" behaviour as other operators may do

Fixes #1334

Author: qwwdfsad

kotlinx-coroutines-core/common/src/channels/Produce.kt

@@ -126,7 +126,7 @@ public fun <E> CoroutineScope.produce(
     return coroutine
 }
 
-internal open class ProducerCoroutine<E>(
+private class ProducerCoroutine<E>(
     parentContext: CoroutineContext, channel: Channel<E>
 ) : ChannelCoroutine<E>(parentContext, channel, active = true), ProducerScope<E> {
     override val isActive: Boolean

kotlinx-coroutines-core/common/src/flow/internal/ChannelFlow.kt

@@ -68,7 +68,7 @@ public abstract class ChannelFlow<T>(
         scope.broadcast(context, produceCapacity, start, block = collectToFun)
 
     open fun produceImpl(scope: CoroutineScope): ReceiveChannel<T> =
-        scope.flowProduce(context, produceCapacity, block = collectToFun)
+        scope.produce(context, produceCapacity, block = collectToFun)
 
     override suspend fun collect(collector: FlowCollector<T>) =
         coroutineScope {

kotlinx-coroutines-core/common/src/flow/internal/FlowCoroutine.kt

@@ -52,20 +52,6 @@ internal fun <R> scopedFlow(@BuilderInference block: suspend CoroutineScope.(Flo
         flowScope { block(collector) }
     }
 
-/*
- * Shortcut for produce { flowScope {block() } }
- */
-internal fun <T> CoroutineScope.flowProduce(
-    context: CoroutineContext,
-    capacity: Int = 0, @BuilderInference block: suspend ProducerScope<T>.() -> Unit
-): ReceiveChannel<T> {
-    val channel = Channel<T>(capacity)
-    val newContext = newCoroutineContext(context)
-    val coroutine = FlowProduceCoroutine(newContext, channel)
-    coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
-    return coroutine
-}
-
 private class FlowCoroutine<T>(
     context: CoroutineContext,
     uCont: Continuation<T>
@@ -75,13 +61,3 @@ private class FlowCoroutine<T>(
         return cancelImpl(cause)
     }
 }
-
-private class FlowProduceCoroutine<T>(
-    parentContext: CoroutineContext,
-    channel: Channel<T>
-) : ProducerCoroutine<T>(parentContext, channel) {
-    public override fun childCancelled(cause: Throwable): Boolean {
-        if (cause is ChildCancelledException) return true
-        return cancelImpl(cause)
-    }
-}

kotlinx-coroutines-core/common/test/flow/channels/ChannelFlowTest.kt

@@ -29,7 +29,6 @@ class ChannelFlowTest : TestBase() {
         assertEquals(listOf(1, 2), flow.toList())
     }
 
-    // todo: this is pretty useless behavior
     @Test
     fun testConflated() = runTest {
         val flow = channelFlow {
@@ -114,6 +113,7 @@ class ChannelFlowTest : TestBase() {
     }
 
     @Test
+    @Ignore // #1374
     fun testBufferWithTimeout() = runTest {
         fun Flow<Int>.bufferWithTimeout(): Flow<Int> = channelFlow {
             expect(2)
@@ -140,4 +140,24 @@ class ChannelFlowTest : TestBase() {
         assertFailsWith<TimeoutCancellationException>(flow)
         finish(6)
     }
+
+    @Test
+    fun testChildCancellation() = runTest {
+        channelFlow {
+            val job = launch {
+                expect(2)
+                hang { expect(4) }
+            }
+            expect(1)
+            yield()
+            expect(3)
+            job.cancelAndJoin()
+            send(5)
+
+        }.collect {
+            expect(it)
+        }
+
+        finish(6)
+    }
 }

kotlinx-coroutines-core/jvm/test/flow/CallbackFlowTest.kt

@@ -83,7 +83,7 @@ class CallbackFlowTest : TestBase() {
             }
         }
 
-        val flow = channelFlow<Int> {
+        val flow = callbackFlow<Int>() {
             api.start(channel)
             awaitClose {
                 api.stop()
@@ -118,7 +118,7 @@ class CallbackFlowTest : TestBase() {
         }
     }
 
-    private fun Flow<Int>.merge(other: Flow<Int>): Flow<Int> = channelFlow {
+    private fun Flow<Int>.merge(other: Flow<Int>): Flow<Int> = callbackFlow {
         launch {
             collect { send(it) }
         }