Get rid of changes in the public API

Author: AlexVanGogen

IntelliJ-patches.md

@@ -77,13 +77,12 @@ Some logic related to instrumentation was extracted to separate methods so that
 - `kotlinx.coroutines.flow.internal.FlowValueWrapperInternalKt.emitInternal(FlowCollector, value)` -- alternative of a regular `FlowCollector.emit` that supports insertion points; if there is a `FlowCollector`, its `emit` call can be replaced with `emitInternal` so this case would also be supported for constructing async stack traces 
 - `kotlinx.coroutines.flow.internal.FlowValueWrapperInternalKt.debuggerCapture` -- common insertion point for a debugger agent; simplifies instrumentation; the value is always being unwrapped inside. `emitInternal` uses this method.
 
+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 `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.
 
-One public method was added to support `buffer` and operators that use it inside:
-- `ReceiveChannel.emitAll`. It encapsulates emitting values in `FlowCollector.emitAllImpl` and has a special implementation in `BufferedChannel`.
-
 Changes were made to lambda parameter `onElementRetrieved` in `BufferedChannel<E>` methods: now they accept `Any?` instead of `E` because now they may be given a wrapped value.
 
 `SelectImplementation.complete` now uses `debuggerCapture` to properly propagate value that might come from flows. 

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

@@ -1552,7 +1552,7 @@ internal open class BufferedChannel<E>(
 
     override fun iterator(): ChannelIterator<E> = BufferedChannelIterator()
 
-    override suspend fun emitAll(collector: FlowCollector<E>) {
+    internal suspend fun emitAllInternal(collector: FlowCollector<E>) {
         val iterator = iterator() as BufferedChannel.BufferedChannelIterator
         while (iterator.hasNext()) {
             collector.emitInternal(iterator.nextInternal())

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

@@ -289,15 +289,6 @@ public interface ReceiveChannel<out E> {
      */
     public operator fun iterator(): ChannelIterator<E>
 
-    /**
-     * Emits all elements of this channel using [collector].
-     */
-    public suspend fun emitAll(collector: FlowCollector<E>) {
-        for (element in this) {
-            collector.emit(element)
-        }
-    }
-
     /**
      * Cancels reception of remaining elements from this channel with an optional [cause].
      * This function closes the channel and removes all buffered sent elements from it.

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 {
-        channel.emitAll(this)
+        emitAll(channel, this)
     } catch (e: Throwable) {
         cause = e
         throw e
@@ -38,6 +38,19 @@ private suspend fun <T> FlowCollector<T>.emitAllImpl(channel: ReceiveChannel<T>,
     }
 }
 
+private suspend fun <T> emitAll(channel: ReceiveChannel<T>, collector: FlowCollector<T>) {
+    when (channel) {
+        is BufferedChannel<*> -> {
+            (channel as BufferedChannel<T>).emitAllInternal(collector)
+        }
+        else -> {
+            for (element in channel) {
+                collector.emit(element)
+            }
+        }
+    }
+}
+
 /**
  * Represents the given receive channel as a hot flow and [receives][ReceiveChannel.receive] from the channel
  * in fan-out fashion every time this flow is collected. One element will be emitted to one collector only.