1.4.0 experimentality

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

@@ -989,7 +989,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/flow/Builders.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/Migration.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(
@@ -482,4 +481,4 @@ public fun <T> Flow<T>.replay(bufferSize: Int): Flow<T> = noImpl()
     message = "Flow analogue of 'cache()' is 'shareIn' with unlimited replay and 'started = SharingStared.Lazily' argument'",
     replaceWith = ReplaceWith("this.shareIn(scope, Int.MAX_VALUE, started = SharingStared.Lazily)")
 )
-public fun <T> Flow<T>.cache(): Flow<T> = noImpl()
+public fun <T> Flow<T>.cache(): Flow<T> = noImpl()

kotlinx-coroutines-core/common/src/flow/operators/Emitters.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
@@ -173,7 +171,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 {
@@ -198,12 +195,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?