Remove experimental status from Flow API (#1963)

* Remove experimental annotation from Flow terminal and Channel operators
* Remove experimental annotation from Flow count* and reduce* operators
* Remove experimental annotation from Flow operations, including buffer and flowOn
* Remove experimental annotation from combine and zip

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

@@ -23,7 +23,6 @@ import kotlinx.coroutines.flow.internal.unsafeFlow as flow
  * This function provides a more efficient shorthand for `channel.consumeEach { value -> emit(value) }`.
  * See [consumeEach][ReceiveChannel.consumeEach].
  */
-@ExperimentalCoroutinesApi // since version 1.3.0
 public suspend fun <T> FlowCollector<T>.emitAll(channel: ReceiveChannel<T>): Unit =
     emitAllImpl(channel, consume = true)
 
@@ -84,7 +83,6 @@ private suspend fun <T> FlowCollector<T>.emitAllImpl(channel: ReceiveChannel<T>,
  * In particular, [produceIn] returns the original channel.
  * Calls to [flowOn] have generally no effect, unless [buffer] is used to explicitly request buffering.
  */
-@ExperimentalCoroutinesApi // since version 1.4.0
 public fun <T> ReceiveChannel<T>.receiveAsFlow(): Flow<T> = ChannelAsFlow(this, consume = false)
 
 /**
@@ -106,7 +104,6 @@ public fun <T> ReceiveChannel<T>.receiveAsFlow(): Flow<T> = ChannelAsFlow(this,
  * In particular, [produceIn] returns the original channel (but throws [IllegalStateException] on repeated calls).
  * Calls to [flowOn] have generally no effect, unless [buffer] is used to explicitly request buffering.
  */
-@ExperimentalCoroutinesApi // since version 1.3.0
 public fun <T> ReceiveChannel<T>.consumeAsFlow(): Flow<T> = ChannelAsFlow(this, consume = true)
 
 /**

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

@@ -104,7 +104,6 @@ import kotlin.jvm.*
  * [RENDEZVOUS][Channel.RENDEZVOUS], [UNLIMITED][Channel.UNLIMITED] or a non-negative value indicating
  * an explicitly requested size.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.buffer(capacity: Int = BUFFERED): Flow<T> {
     require(capacity >= 0 || capacity == BUFFERED || capacity == CONFLATED) {
         "Buffer size should be non-negative, BUFFERED, or CONFLATED, but was $capacity"
@@ -147,7 +146,6 @@ public fun <T> Flow<T>.buffer(capacity: Int = BUFFERED): Flow<T> {
  * always fused so that only one properly configured channel is used for execution.
  * **Conflation takes precedence over `buffer()` calls with any other capacity.**
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.conflate(): Flow<T> = buffer(CONFLATED)
 
 /**
@@ -194,7 +192,6 @@ public fun <T> Flow<T>.conflate(): Flow<T> = buffer(CONFLATED)
  *
  * @throws [IllegalArgumentException] if provided context contains [Job] instance.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.flowOn(context: CoroutineContext): Flow<T> {
     checkFlowContext(context)
     return when {

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

@@ -15,22 +15,19 @@ import kotlinx.coroutines.flow.internal.unsafeFlow as flow
 /**
  * Returns flow where all subsequent repetitions of the same value are filtered out.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.distinctUntilChanged(): Flow<T> = distinctUntilChangedBy { it }
 
 /**
  * Returns flow where all subsequent repetitions of the same value are filtered out, when compared
  * with each other via the provided [areEquivalent] function.
  */
-@FlowPreview
 public fun <T> Flow<T>.distinctUntilChanged(areEquivalent: (old: T, new: T) -> Boolean): Flow<T> =
     distinctUntilChangedBy(keySelector = { it }, areEquivalent = areEquivalent)
 
 /**
  * Returns flow where all subsequent repetitions of the same key are filtered out, where
  * key is extracted with [keySelector] function.
  */
-@FlowPreview
 public fun <T, K> Flow<T>.distinctUntilChangedBy(keySelector: (T) -> K): Flow<T> =
     distinctUntilChangedBy(keySelector = keySelector, areEquivalent = { old, new -> old == new })
 

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

@@ -34,7 +34,6 @@ import kotlin.jvm.*
  * }
  * ```
  */
-@ExperimentalCoroutinesApi
 public inline fun <T, R> Flow<T>.transform(
     @BuilderInference crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
 ): Flow<R> = flow { // Note: safe flow is used here, because collector is exposed to transform on each operation

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

@@ -54,7 +54,6 @@ import kotlinx.coroutines.flow.internal.unsafeFlow as flow
  * retry an original flow use [retryWhen] operator that can retry the flow multiple times without
  * introducing ever-growing stack of suspending calls.
  */
-@ExperimentalCoroutinesApi // tentatively stable in 1.3.0
 public fun <T> Flow<T>.catch(action: suspend FlowCollector<T>.(cause: Throwable) -> Unit): Flow<T> =
     flow {
         val exception = catchImpl(this)
@@ -117,7 +116,6 @@ public fun <T> Flow<T>.onErrorCollect(
  *
  * @throws IllegalArgumentException when [retries] is not positive.
  */
-@ExperimentalCoroutinesApi // tentatively stable in 1.3.0
 public fun <T> Flow<T>.retry(
     retries: Long = Long.MAX_VALUE,
     predicate: suspend (cause: Throwable) -> Boolean = { true }
@@ -169,7 +167,6 @@ public fun <T> Flow<T>.retry(
  *
  * See [catch] for more details.
  */
-@ExperimentalCoroutinesApi // tentatively stable in 1.3.0
 public fun <T> Flow<T>.retryWhen(predicate: suspend FlowCollector<T>.(cause: Throwable, attempt: Long) -> Boolean): Flow<T> =
     flow {
         var attempt = 0L

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

@@ -7,7 +7,6 @@
 
 package kotlinx.coroutines.flow
 
-import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.internal.*
 import kotlin.jvm.*
 import kotlinx.coroutines.flow.internal.unsafeFlow as flow
@@ -16,7 +15,6 @@ import kotlinx.coroutines.flow.internal.unsafeFlow as flow
  * Returns a flow that ignores first [count] elements.
  * Throws [IllegalArgumentException] if [count] is negative.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.drop(count: Int): Flow<T> {
     require(count >= 0) { "Drop count should be non-negative, but had $count" }
     return flow {
@@ -30,7 +28,6 @@ public fun <T> Flow<T>.drop(count: Int): Flow<T> {
 /**
  * Returns a flow containing all elements except first elements that satisfy the given predicate.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.dropWhile(predicate: suspend (T) -> Boolean): Flow<T> = flow {
     var matched = false
     collect { value ->
@@ -48,7 +45,6 @@ public fun <T> Flow<T>.dropWhile(predicate: suspend (T) -> Boolean): Flow<T> = f
  * When [count] elements are consumed, the original flow is cancelled.
  * Throws [IllegalArgumentException] if [count] is not positive.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.take(count: Int): Flow<T> {
     require(count > 0) { "Requested element count $count should be positive" }
     return flow {
@@ -75,7 +71,6 @@ private suspend fun <T> FlowCollector<T>.emitAbort(value: T) {
 /**
  * Returns a flow that contains first elements satisfying the given [predicate].
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.takeWhile(predicate: suspend (T) -> Boolean): Flow<T> = flow {
     try {
         collect { value ->

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

@@ -59,7 +59,6 @@ public inline fun <T, R: Any> Flow<T>.mapNotNull(crossinline transform: suspend
 /**
  * Returns a flow that wraps each element into [IndexedValue], containing value and its index (starting from zero).
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.withIndex(): Flow<IndexedValue<T>> = flow {
     var index = 0
     collect { value ->

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

@@ -30,7 +30,6 @@ import kotlinx.coroutines.flow.internal.unsafeFlow as flow
  * This function is a shorthand for `flow.combineTransform(flow2) { a, b -> emit(transform(a, b)) }
  */
 @JvmName("flowCombine")
-@ExperimentalCoroutinesApi
 public fun <T1, T2, R> Flow<T1>.combine(flow: Flow<T2>, transform: suspend (a: T1, b: T2) -> R): Flow<R> = flow {
     combineTransformInternal(this@combine, flow) { a, b ->
         emit(transform(a, b))
@@ -52,7 +51,6 @@ public fun <T1, T2, R> Flow<T1>.combine(flow: Flow<T2>, transform: suspend (a: T
  *
  * This function is a shorthand for `combineTransform(flow, flow2) { a, b -> emit(transform(a, b)) }
  */
-@ExperimentalCoroutinesApi
 public fun <T1, T2, R> combine(flow: Flow<T1>, flow2: Flow<T2>, transform: suspend (a: T1, b: T2) -> R): Flow<R> =
     flow.combine(flow2, transform)
 
@@ -74,7 +72,6 @@ public fun <T1, T2, R> combine(flow: Flow<T1>, flow2: Flow<T2>, transform: suspe
  * ```
  */
 @JvmName("flowCombineTransform")
-@ExperimentalCoroutinesApi
 public fun <T1, T2, R> Flow<T1>.combineTransform(
     flow: Flow<T2>,
     @BuilderInference transform: suspend FlowCollector<R>.(a: T1, b: T2) -> Unit
@@ -101,7 +98,6 @@ public fun <T1, T2, R> Flow<T1>.combineTransform(
  * }
  * ```
  */
-@ExperimentalCoroutinesApi
 public fun <T1, T2, R> combineTransform(
     flow: Flow<T1>,
     flow2: Flow<T2>,
@@ -117,7 +113,6 @@ public fun <T1, T2, R> combineTransform(
  * Returns a [Flow] whose values are generated with [transform] function by combining
  * the most recently emitted values by each flow.
  */
-@ExperimentalCoroutinesApi
 public inline fun <T1, T2, T3, R> combine(
     flow: Flow<T1>,
     flow2: Flow<T2>,
@@ -137,7 +132,6 @@ public inline fun <T1, T2, T3, R> combine(
  * The receiver of the [transform] is [FlowCollector] and thus `transform` is a
  * generic function that may transform emitted element, skip it or emit it multiple times.
  */
-@ExperimentalCoroutinesApi
 public inline fun <T1, T2, T3, R> combineTransform(
     flow: Flow<T1>,
     flow2: Flow<T2>,
@@ -155,7 +149,6 @@ public inline fun <T1, T2, T3, R> combineTransform(
  * Returns a [Flow] whose values are generated with [transform] function by combining
  * the most recently emitted values by each flow.
  */
-@ExperimentalCoroutinesApi
 public inline fun <T1, T2, T3, T4, R> combine(
     flow: Flow<T1>,
     flow2: Flow<T2>,
@@ -177,7 +170,6 @@ public inline fun <T1, T2, T3, T4, R> combine(
  * The receiver of the [transform] is [FlowCollector] and thus `transform` is a
  * generic function that may transform emitted element, skip it or emit it multiple times.
  */
-@ExperimentalCoroutinesApi
 public inline fun <T1, T2, T3, T4, R> combineTransform(
     flow: Flow<T1>,
     flow2: Flow<T2>,
@@ -197,7 +189,6 @@ public inline fun <T1, T2, T3, T4, R> combineTransform(
  * Returns a [Flow] whose values are generated with [transform] function by combining
  * the most recently emitted values by each flow.
  */
-@ExperimentalCoroutinesApi
 public inline fun <T1, T2, T3, T4, T5, R> combine(
     flow: Flow<T1>,
     flow2: Flow<T2>,
@@ -221,7 +212,6 @@ public inline fun <T1, T2, T3, T4, T5, R> combine(
  * The receiver of the [transform] is [FlowCollector] and thus `transform` is a
  * generic function that may transform emitted element, skip it or emit it multiple times.
  */
-@ExperimentalCoroutinesApi
 public inline fun <T1, T2, T3, T4, T5, R> combineTransform(
     flow: Flow<T1>,
     flow2: Flow<T2>,
@@ -243,7 +233,6 @@ public inline fun <T1, T2, T3, T4, T5, R> combineTransform(
  * Returns a [Flow] whose values are generated with [transform] function by combining
  * the most recently emitted values by each flow.
  */
-@ExperimentalCoroutinesApi
 public inline fun <reified T, R> combine(
     vararg flows: Flow<T>,
     crossinline transform: suspend (Array<T>) -> R
@@ -257,7 +246,6 @@ public inline fun <reified T, R> combine(
  * The receiver of the [transform] is [FlowCollector] and thus `transform` is a
  * generic function that may transform emitted element, skip it or emit it multiple times.
  */
-@ExperimentalCoroutinesApi
 public inline fun <reified T, R> combineTransform(
     vararg flows: Flow<T>,
     @BuilderInference crossinline transform: suspend FlowCollector<R>.(Array<T>) -> Unit
@@ -269,7 +257,6 @@ public inline fun <reified T, R> combineTransform(
  * Returns a [Flow] whose values are generated with [transform] function by combining
  * the most recently emitted values by each flow.
  */
-@ExperimentalCoroutinesApi
 public inline fun <reified T, R> combine(
     flows: Iterable<Flow<T>>,
     crossinline transform: suspend (Array<T>) -> R
@@ -289,7 +276,6 @@ public inline fun <reified T, R> combine(
  * The receiver of the [transform] is [FlowCollector] and thus `transform` is a
  * generic function that may transform emitted element, skip it or emit it multiple times.
  */
-@ExperimentalCoroutinesApi
 public inline fun <reified T, R> combineTransform(
     flows: Iterable<Flow<T>>,
     @BuilderInference crossinline transform: suspend FlowCollector<R>.(Array<T>) -> Unit
@@ -313,5 +299,4 @@ public inline fun <reified T, R> combineTransform(
  * }
  * ```
  */
-@ExperimentalCoroutinesApi
 public fun <T1, T2, R> Flow<T1>.zip(other: Flow<T2>, transform: suspend (T1, T2) -> R): Flow<R> = zipImpl(this, other, transform)

kotlinx-coroutines-core/common/src/flow/terminal/Collect.kt

@@ -46,7 +46,6 @@ public suspend fun Flow<*>.collect(): Unit = collect(NopCollector)
  *
  * Note that resulting value of [launchIn] is not used the provided scope takes care of cancellation.
  */
-@ExperimentalCoroutinesApi // tentatively stable in 1.3.0
 public fun <T> Flow<T>.launchIn(scope: CoroutineScope): Job = scope.launch {
     collect() // tail-call
 }
@@ -80,7 +79,6 @@ public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value
  *
  * See also [collect] and [withIndex].
  */
-@ExperimentalCoroutinesApi
 public suspend inline fun <T> Flow<T>.collectIndexed(crossinline action: suspend (index: Int, value: T) -> Unit): Unit =
     collect(object : FlowCollector<T> {
         private var index = 0
@@ -108,7 +106,6 @@ public suspend inline fun <T> Flow<T>.collectIndexed(crossinline action: suspend
  *
  * prints "Collecting 1, Collecting 2, 2 collected"
  */
-@ExperimentalCoroutinesApi
 public suspend fun <T> Flow<T>.collectLatest(action: suspend (value: T) -> Unit) {
     /*
      * Implementation note:
@@ -131,5 +128,4 @@ public suspend fun <T> Flow<T>.collectLatest(action: suspend (value: T) -> Unit)
  * It is a shorthand for `flow.collect { value -> emit(value) }`.
  */
 @BuilderInference
-@ExperimentalCoroutinesApi
 public suspend inline fun <T> FlowCollector<T>.emitAll(flow: Flow<T>): Unit = flow.collect(this)

kotlinx-coroutines-core/common/src/flow/terminal/Count.kt

@@ -13,7 +13,6 @@ import kotlin.jvm.*
 /**
  * Returns the number of elements in this flow.
  */
-@ExperimentalCoroutinesApi
 public suspend fun <T> Flow<T>.count(): Int  {
     var i = 0
     collect {
@@ -26,7 +25,6 @@ public suspend fun <T> Flow<T>.count(): Int  {
 /**
  * Returns the number of elements matching the given predicate.
  */
-@ExperimentalCoroutinesApi
 public suspend fun <T> Flow<T>.count(predicate: suspend (T) -> Boolean): Int {
     var i = 0
     collect { value ->

kotlinx-coroutines-core/common/src/flow/terminal/Reduce.kt

@@ -17,7 +17,6 @@ import kotlin.jvm.*
  * Accumulates value starting with the first element and applying [operation] to current accumulator value and each element.
  * Throws [NoSuchElementException] if flow was empty.
  */
-@ExperimentalCoroutinesApi
 public suspend fun <S, T : S> Flow<T>.reduce(operation: suspend (accumulator: S, value: T) -> S): S {
     var accumulator: Any? = NULL
 
@@ -38,7 +37,6 @@ public suspend fun <S, T : S> Flow<T>.reduce(operation: suspend (accumulator: S,
 /**
  * Accumulates value starting with [initial] value and applying [operation] current accumulator value and each element
  */
-@ExperimentalCoroutinesApi
 public suspend inline fun <T, R> Flow<T>.fold(
     initial: R,
     crossinline operation: suspend (acc: R, value: T) -> R