Get rid of changes in the public API -- bring back CoroutineChannel support

Author: AlexVanGogen

IntelliJ-patches.md

@@ -79,6 +79,10 @@ Some logic related to instrumentation was extracted to separate methods so that
 
 One internal method was added to `BufferedChannel`: `emitAllInternal`. This method ensures the value will be unwrapped in an insertion point.
 
+One internal method was added to `flow/Channels.kt`: `emitAllInternal`. It emits all values, like usual, but also considers wrapping/unwrapping supported in `BufferedChannel`.
+
+One internal method was added to `ChannelCoroutine`: `emitAllInternal` serves to bridge its delegate and the method above.
+
 One internal method was added to `BufferedChannelIterator`: `nextInternal` -- same as `next` but may return a wrapped value. It should only be used with a function that is capable of unwrapping the value (see `BufferedChannel.emitAll` and `BufferedChannelIterator.next`), so there's a guarantee a wrapped value will always unwrap before emitting.
 
 Why not just let `next` return a maybe wrapped value? That's because it is heavily used outside a currently supported scope. For example, one may just indirectly call it from a for-loop. In this case, unwrapping will never happen, and a user will get a handful of `ClassCastException`s.

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

@@ -1,6 +1,8 @@
 package kotlinx.coroutines.channels
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.coroutines.flow.emitAllInternal
 import kotlin.coroutines.*
 
 internal open class ChannelCoroutine<E>(
@@ -35,4 +37,8 @@ internal open class ChannelCoroutine<E>(
         _channel.cancel(exception) // cancel the channel
         cancelCoroutine(exception) // cancel the job
     }
+
+    internal suspend fun emitAllInternal(flowCollector: FlowCollector<E>) {
+        emitAllInternal(_channel, flowCollector)
+    }
 }

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

@@ -29,7 +29,7 @@ private suspend fun <T> FlowCollector<T>.emitAllImpl(channel: ReceiveChannel<T>,
     ensureActive()
     var cause: Throwable? = null
     try {
-        emitAll(channel, this)
+        emitAllInternal(channel, this)
     } catch (e: Throwable) {
         cause = e
         throw e
@@ -38,11 +38,14 @@ private suspend fun <T> FlowCollector<T>.emitAllImpl(channel: ReceiveChannel<T>,
     }
 }
 
-private suspend fun <T> emitAll(channel: ReceiveChannel<T>, collector: FlowCollector<T>) {
+internal suspend fun <T> emitAllInternal(channel: ReceiveChannel<T>, collector: FlowCollector<T>) {
     when (channel) {
         is BufferedChannel<*> -> {
             (channel as BufferedChannel<T>).emitAllInternal(collector)
         }
+        is ChannelCoroutine<*> -> {
+            (channel as ChannelCoroutine<T>).emitAllInternal(collector)
+        }
         else -> {
             for (element in channel) {
                 collector.emit(element)