Update experimental declarations (Kotlin#2316)

* Gracefully increase deprecation level on Channel operators instead of removing them, a warning was not strict enough
* Remove hidden onCompletion from -M release
* Promote StateFlow and SharedFlow to stable API
* Lift out experimentality where it is applicable

    * CoroutineDispatcher.invoke
    * ReceiveChannel.consume and ReceiveChannel.consumeEach
    * Flow core operators: onStart, onCompletion, onEmpty
    * CompletableDeferred.completeWith
    * awaitCancellation
    * Add experimentality notes where applicable

Author: qwwdfsad

benchmarks/src/jmh/kotlin/benchmarks/ChannelSinkBenchmark.kt

@@ -50,4 +50,22 @@ open class ChannelSinkBenchmark {
         for (i in start until (start + count))
             send(i)
     }
+
+    // Migrated from deprecated operators, are good only for stressing channels
+
+    private fun <E> ReceiveChannel<E>.filter(context: CoroutineContext = Dispatchers.Unconfined, predicate: suspend (E) -> Boolean): ReceiveChannel<E> =
+        GlobalScope.produce(context, onCompletion = { cancel() }) {
+            for (e in this@filter) {
+                if (predicate(e)) send(e)
+            }
+        }
+
+    private suspend inline fun <E, R> ReceiveChannel<E>.fold(initial: R, operation: (acc: R, E) -> R): R {
+        var accumulator = initial
+        consumeEach {
+            accumulator = operation(accumulator, it)
+        }
+        return accumulator
+    }
 }
+

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

@@ -992,7 +992,6 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun merge (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun merge ([Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun observeOn (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/CoroutineContext;)Lkotlinx/coroutines/flow/Flow;
-	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;

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

@@ -175,7 +175,6 @@ public suspend fun <T> withContext(
  *
  * This inline function calls [withContext].
  */
-@ExperimentalCoroutinesApi
 public suspend inline operator fun <T> CoroutineDispatcher.invoke(
     noinline block: suspend CoroutineScope.() -> T
 ): T = withContext(this, block)

kotlinx-coroutines-core/common/src/CompletableDeferred.kt

@@ -57,7 +57,6 @@ public interface CompletableDeferred<T> : Deferred<T> {
  * This function transitions this deferred in the same ways described by [CompletableDeferred.complete] and
  * [CompletableDeferred.completeExceptionally].
  */
-@ExperimentalCoroutinesApi // since 1.3.2, tentatively until 1.4.0
 public fun <T> CompletableDeferred<T>.completeWith(result: Result<T>): Boolean =
     result.fold({ complete(it) }, { completeExceptionally(it) })
 

kotlinx-coroutines-core/common/src/CoroutineStart.kt

@@ -55,7 +55,7 @@ public enum class CoroutineStart {
      * Cancellability of coroutine at suspension points depends on the particular implementation details of
      * suspending functions as in [DEFAULT].
      */
-    @ExperimentalCoroutinesApi
+    @ExperimentalCoroutinesApi // Since 1.0.0, no ETA on stability
     ATOMIC,
 
     /**
@@ -71,7 +71,7 @@ public enum class CoroutineStart {
      *
      * **Note: This is an experimental api.** Execution semantics of coroutines may change in the future when this mode is used.
      */
-    @ExperimentalCoroutinesApi
+    @ExperimentalCoroutinesApi  // Since 1.0.0, no ETA on stability
     UNDISPATCHED;
 
     /**

kotlinx-coroutines-core/common/src/Debug.common.kt

@@ -27,7 +27,7 @@ internal expect fun assert(value: () -> Boolean)
  * Copy mechanism is used only on JVM, but it might be convenient to implement it in common exceptions,
  * so on JVM their stacktraces will be properly recovered.
  */
-@ExperimentalCoroutinesApi
+@ExperimentalCoroutinesApi // Since 1.2.0, no ETA on stability
 public interface CopyableThrowable<T> where T : Throwable, T : CopyableThrowable<T> {
 
     /**

kotlinx-coroutines-core/common/src/Delay.kt

@@ -95,7 +95,6 @@ public interface Delay {
  * }
  * ```
  */
-@ExperimentalCoroutinesApi
 public suspend fun awaitCancellation(): Nothing = suspendCancellableCoroutine {}
 
 /**

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

@@ -47,7 +47,7 @@ public fun <E> ReceiveChannel<E>.broadcast(
     val scope = GlobalScope + Dispatchers.Unconfined + CoroutineExceptionHandler { _, _ -> }
     // We can run this coroutine in the context that ignores all exceptions, because of `onCompletion = consume()`
     // which passes all exceptions upstream to the source ReceiveChannel
-    return scope.broadcast(capacity = capacity, start = start, onCompletion = consumes()) {
+    return scope.broadcast(capacity = capacity, start = start, onCompletion = { cancelConsumed(it) }) {
         for (e in this@broadcast) {
             send(e)
         }

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

@@ -16,7 +16,6 @@ import kotlinx.coroutines.*
  * * [DROP_LATEST] — drop **the latest** value that is being added to the buffer right now on buffer overflow
  *   (so that buffer contents stay the same), do not suspend.
  */
-@ExperimentalCoroutinesApi
 public enum class BufferOverflow {
     /**
      * Suspend on buffer overflow.

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

@@ -204,8 +204,8 @@ public fun LongRange.asFlow(): Flow<Long> = flow {
 @FlowPreview
 @Deprecated(
     message = "Use channelFlow with awaitClose { } instead of flowViaChannel and invokeOnClose { }.",
-    level = DeprecationLevel.WARNING
-)
+    level = DeprecationLevel.ERROR
+) // To be removed in 1.4.x
 @Suppress("DeprecatedCallableAddReplaceWith")
 public fun <T> flowViaChannel(
     bufferSize: Int = BUFFERED,

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

@@ -434,7 +434,6 @@ public fun <T, R> Flow<T>.switchMap(transform: suspend (value: T) -> Flow<R>): F
     message = "'scanReduce' was renamed to 'runningReduce' to be consistent with Kotlin standard library",
     replaceWith = ReplaceWith("runningReduce(operation)")
 )
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.scanReduce(operation: suspend (accumulator: T, value: T) -> T): Flow<T> = runningReduce(operation)
 
 @Deprecated(

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

@@ -108,7 +108,6 @@ import kotlin.native.concurrent.*
  * might be added to this interface in the future, but is stable for use.
  * Use the `MutableSharedFlow(replay, ...)` constructor function to create an implementation.
  */
-@ExperimentalCoroutinesApi
 public interface SharedFlow<out T> : Flow<T> {
     /**
      * A snapshot of the replay cache.
@@ -138,7 +137,6 @@ public interface SharedFlow<out T> : Flow<T> {
  * might be added to this interface in the future, but is stable for use.
  * Use the `MutableSharedFlow(...)` constructor function to create an implementation.
  */
-@ExperimentalCoroutinesApi
 public interface MutableSharedFlow<T> : SharedFlow<T>, FlowCollector<T> {
     /**
      * Tries to emit a [value] to this shared flow without suspending. It returns `true` if the value was
@@ -202,7 +200,6 @@ public interface MutableSharedFlow<T> : SharedFlow<T>, FlowCollector<T> {
  *   supported only when `replay > 0` or `extraBufferCapacity > 0`).
  */
 @Suppress("FunctionName", "UNCHECKED_CAST")
-@ExperimentalCoroutinesApi
 public fun <T> MutableSharedFlow(
     replay: Int = 0,
     extraBufferCapacity: Int = 0,

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

@@ -12,7 +12,6 @@ import kotlin.time.*
  * A command emitted by [SharingStarted] implementations to control the sharing coroutine in
  * the [shareIn] and [stateIn] operators.
  */
-@ExperimentalCoroutinesApi
 public enum class SharingCommand {
     /**
      * Starts sharing, launching collection of the upstream flow.
@@ -75,19 +74,16 @@ public enum class SharingCommand {
  * The completion of the `command` flow normally has no effect (the upstream flow keeps running if it was running).
  * The failure of the `command` flow cancels the sharing coroutine and the upstream flow.
  */
-@ExperimentalCoroutinesApi
 public interface SharingStarted {
     public companion object {
         /**
          * Sharing is started immediately and never stops.
          */
-        @ExperimentalCoroutinesApi
         public val Eagerly: SharingStarted = StartedEagerly()
 
         /**
          * Sharing is started when the first subscriber appears and never stops.
          */
-        @ExperimentalCoroutinesApi
         public val Lazily: SharingStarted = StartedLazily()
 
         /**
@@ -108,7 +104,6 @@ public interface SharingStarted {
          * are negative.
          */
         @Suppress("FunctionName")
-        @ExperimentalCoroutinesApi
         public fun WhileSubscribed(
             stopTimeoutMillis: Long = 0,
             replayExpirationMillis: Long = Long.MAX_VALUE
@@ -143,7 +138,6 @@ public interface SharingStarted {
  */
 @Suppress("FunctionName")
 @ExperimentalTime
-@ExperimentalCoroutinesApi
 public fun SharingStarted.Companion.WhileSubscribed(
     stopTimeout: Duration = Duration.ZERO,
     replayExpiration: Duration = Duration.INFINITE

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

@@ -135,7 +135,6 @@ import kotlin.native.concurrent.*
  * might be added to this interface in the future, but is stable for use.
  * Use the `MutableStateFlow(value)` constructor function to create an implementation.
  */
-@ExperimentalCoroutinesApi
 public interface StateFlow<out T> : SharedFlow<T> {
     /**
      * The current value of this state flow.
@@ -156,7 +155,6 @@ public interface StateFlow<out T> : SharedFlow<T> {
  * might be added to this interface in the future, but is stable for use.
  * Use the `MutableStateFlow()` constructor function to create an implementation.
  */
-@ExperimentalCoroutinesApi
 public interface MutableStateFlow<T> : StateFlow<T>, MutableSharedFlow<T> {
     /**
      * The current value of this state flow.
@@ -180,7 +178,6 @@ public interface MutableStateFlow<T> : StateFlow<T>, MutableSharedFlow<T> {
  * Creates a [MutableStateFlow] with the given initial [value].
  */
 @Suppress("FunctionName")
-@ExperimentalCoroutinesApi
 public fun <T> MutableStateFlow(value: T): MutableStateFlow<T> = StateFlowImpl(value ?: NULL)
 
 // ------------------------------------ Implementation ------------------------------------
@@ -380,4 +377,4 @@ internal fun <T> StateFlow<T>.fuseStateFlow(
         return this
     }
     return fuseSharedFlow(context, capacity, onBufferOverflow)
-}
+}

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

@@ -71,7 +71,6 @@ internal inline fun <T, R> Flow<T>.unsafeTransform(
  *     .collect { println(it) } // prints Begin, a, b, c
  * ```
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.onStart(
     action: suspend FlowCollector<T>.() -> Unit
 ): Flow<T> = unsafeFlow { // Note: unsafe flow is used here, but safe collector is used to invoke start action
@@ -142,7 +141,6 @@ public fun <T> Flow<T>.onStart(
  * In case of failure or cancellation, any attempt to emit additional elements throws the corresponding exception.
  * Use [catch] if you need to suppress failure and replace it with emission of elements.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.onCompletion(
     action: suspend FlowCollector<T>.(cause: Throwable?) -> Unit
 ): Flow<T> = unsafeFlow { // Note: unsafe flow is used here, but safe collector is used to invoke completion action
@@ -178,7 +176,6 @@ public fun <T> Flow<T>.onCompletion(
  * }.collect { println(it) } // prints 1, 2
  * ```
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.onEmpty(
     action: suspend FlowCollector<T>.() -> Unit
 ): Flow<T> = unsafeFlow {
@@ -203,12 +200,6 @@ private class ThrowingCollector(private val e: Throwable) : FlowCollector<Any?>
     }
 }
 
-// It was only released in 1.3.0-M2, remove in 1.4.0
-/** @suppress */
-@Deprecated(level = DeprecationLevel.HIDDEN, message = "binary compatibility with a version w/o FlowCollector receiver")
-public fun <T> Flow<T>.onCompletion(action: suspend (cause: Throwable?) -> Unit): Flow<T> =
-    onCompletion { action(it) }
-
 private suspend fun <T> FlowCollector<T>.invokeSafely(
     action: suspend FlowCollector<T>.(cause: Throwable?) -> Unit,
     cause: Throwable?

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

@@ -132,7 +132,6 @@ import kotlin.jvm.*
  * @param started the strategy that controls when sharing is started and stopped.
  * @param replay the number of values replayed to new subscribers (cannot be negative, defaults to zero).
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.shareIn(
     scope: CoroutineScope,
     started: SharingStarted,
@@ -297,7 +296,6 @@ private fun <T> CoroutineScope.launchSharing(
  *   This value is also used when the state flow is reset using the [SharingStarted.WhileSubscribed] strategy
  *   with the `replayExpirationMillis` parameter.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.stateIn(
     scope: CoroutineScope,
     started: SharingStarted,
@@ -316,7 +314,6 @@ public fun <T> Flow<T>.stateIn(
  *
  * @param scope the coroutine scope in which sharing is started.
  */
-@ExperimentalCoroutinesApi
 public suspend fun <T> Flow<T>.stateIn(scope: CoroutineScope): StateFlow<T> {
     val config = configureSharing(1)
     val result = CompletableDeferred<StateFlow<T>>()
@@ -353,14 +350,12 @@ private fun <T> CoroutineScope.launchSharingDeferred(
 /**
  * Represents this mutable shared flow as a read-only shared flow.
  */
-@ExperimentalCoroutinesApi
 public fun <T> MutableSharedFlow<T>.asSharedFlow(): SharedFlow<T> =
     ReadonlySharedFlow(this)
 
 /**
  * Represents this mutable state flow as a read-only state flow.
  */
-@ExperimentalCoroutinesApi
 public fun <T> MutableStateFlow<T>.asStateFlow(): StateFlow<T> =
     ReadonlyStateFlow(this)
 
@@ -391,7 +386,6 @@ private class ReadonlyStateFlow<T>(
  *
  * The receiver of the [action] is [FlowCollector], so `onSubscription` can emit additional elements.
  */
-@ExperimentalCoroutinesApi
 public fun <T> SharedFlow<T>.onSubscription(action: suspend FlowCollector<T>.() -> Unit): SharedFlow<T> =
     SubscribedSharedFlow(this, action)
 

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

@@ -163,7 +163,7 @@ class ProduceTest : TestBase() {
     private suspend fun cancelOnCompletion(coroutineContext: CoroutineContext) = CoroutineScope(coroutineContext).apply {
         val source = Channel<Int>()
         expect(1)
-        val produced = produce<Int>(coroutineContext, onCompletion = source.consumes()) {
+        val produced = produce<Int>(coroutineContext, onCompletion = { source.cancelConsumed(it) }) {
             expect(2)
             source.receive()
         }

kotlinx-coroutines-core/common/test/channels/ChannelsTest.kt

@@ -14,7 +14,9 @@ class ChannelsJvmTest : TestBase() {
     fun testBlocking() {
         val ch = Channel<Int>()
         val sum = GlobalScope.async {
-            ch.sumBy { it }
+            var sum = 0
+            ch.consumeEach { sum += it }
+            sum
         }
         repeat(10) {
             ch.sendBlocking(it)

kotlinx-coroutines-core/common/test/channels/ProduceTest.kt

@@ -28,7 +28,7 @@ import org.reactivestreams.FlowAdapters
  * **Note: This is an experimental api.** Behaviour of publishers that work as children in a parent scope with respect
  *        to cancellation and error handling may change in the future.
  */
-@ExperimentalCoroutinesApi
+@ExperimentalCoroutinesApi // Since 1.3.x
 public fun <T> flowPublish(
     context: CoroutineContext = EmptyCoroutineContext,
     @BuilderInference block: suspend ProducerScope<T>.() -> Unit

kotlinx-coroutines-core/jvm/test/channels/ChannelsConsumeTest.kt

@@ -24,7 +24,7 @@ import kotlinx.coroutines.flow.*
  * Adjacent applications of [flowOn], [buffer], [conflate], and [produceIn] to the result of `asFlow` are fused.
  * [conflate] has no effect, as this flow is already conflated; one can use [buffer] to change that instead.
  */
-@ExperimentalCoroutinesApi
+@ExperimentalCoroutinesApi // Since 1.3.x
 public fun <T> ObservableValue<T>.asFlow(): Flow<T> = callbackFlow<T> {
     val listener = ChangeListener<T> { _, _, newValue ->
         try {