Promote the bare minimum Flow API to stable for incoming 1.3.0-RC

  * Extract SafeFlow for nicer stacktraces

Author: qwwdfsad

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

@@ -44,12 +44,12 @@ import kotlin.jvm.*
  * ```
  * If you want to switch the context of execution of a flow, use the [flowOn] operator.
  */
-@ExperimentalCoroutinesApi
-public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> {
-    return object : Flow<T> {
-        override suspend fun collect(collector: FlowCollector<T>) {
-            SafeCollector(collector, coroutineContext).block()
-        }
+public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)
+
+// Named anonymous object
+private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : Flow<T> {
+    override suspend fun collect(collector: FlowCollector<T>) {
+        SafeCollector(collector, coroutineContext).block()
     }
 }
 
@@ -90,7 +90,6 @@ public fun <T> (suspend () -> T).asFlow(): Flow<T> = unsafeFlow {
 /**
  * Creates a flow that produces values from the given iterable.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Iterable<T>.asFlow(): Flow<T> = unsafeFlow {
     forEach { value ->
         emit(value)
@@ -100,7 +99,6 @@ public fun <T> Iterable<T>.asFlow(): Flow<T> = unsafeFlow {
 /**
  * Creates a flow that produces values from the given iterable.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Iterator<T>.asFlow(): Flow<T> = unsafeFlow {
     forEach { value ->
         emit(value)
@@ -110,7 +108,6 @@ public fun <T> Iterator<T>.asFlow(): Flow<T> = unsafeFlow {
 /**
  * Creates a flow that produces values from the given sequence.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Sequence<T>.asFlow(): Flow<T> = unsafeFlow {
     forEach { value ->
         emit(value)
@@ -120,7 +117,6 @@ public fun <T> Sequence<T>.asFlow(): Flow<T> = unsafeFlow {
 /**
  * Creates a flow that produces values from the given array of elements.
  */
-@ExperimentalCoroutinesApi
 public fun <T> flowOf(vararg elements: T): Flow<T> = unsafeFlow {
     for (element in elements) {
         emit(element)
@@ -130,7 +126,6 @@ public fun <T> flowOf(vararg elements: T): Flow<T> = unsafeFlow {
 /**
  * Creates flow that produces a given [value].
  */
-@ExperimentalCoroutinesApi
 public fun <T> flowOf(value: T): Flow<T> = unsafeFlow {
     /*
      * Implementation note: this is just an "optimized" overload of flowOf(vararg)
@@ -142,7 +137,6 @@ public fun <T> flowOf(value: T): Flow<T> = unsafeFlow {
 /**
  * Returns an empty flow.
  */
-@ExperimentalCoroutinesApi
 public fun <T> emptyFlow(): Flow<T> = EmptyFlow
 
 private object EmptyFlow : Flow<Nothing> {
@@ -152,7 +146,6 @@ private object EmptyFlow : Flow<Nothing> {
 /**
  * Creates a flow that produces values from the given array.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Array<T>.asFlow(): Flow<T> = unsafeFlow {
     forEach { value ->
         emit(value)
@@ -162,7 +155,6 @@ public fun <T> Array<T>.asFlow(): Flow<T> = unsafeFlow {
 /**
  * Creates flow that produces values from the given array.
  */
-@ExperimentalCoroutinesApi
 public fun IntArray.asFlow(): Flow<Int> = unsafeFlow {
     forEach { value ->
         emit(value)
@@ -172,7 +164,6 @@ public fun IntArray.asFlow(): Flow<Int> = unsafeFlow {
 /**
  * Creates flow that produces values from the given array.
  */
-@ExperimentalCoroutinesApi
 public fun LongArray.asFlow(): Flow<Long> = unsafeFlow {
     forEach { value ->
         emit(value)
@@ -182,7 +173,6 @@ public fun LongArray.asFlow(): Flow<Long> = unsafeFlow {
 /**
  * Creates flow that produces values from the given range.
  */
-@ExperimentalCoroutinesApi
 public fun IntRange.asFlow(): Flow<Int> = unsafeFlow {
     forEach { value ->
         emit(value)
@@ -192,7 +182,6 @@ public fun IntRange.asFlow(): Flow<Int> = unsafeFlow {
 /**
  * Creates flow that produces values from the given range.
  */
-@ExperimentalCoroutinesApi
 public fun LongRange.asFlow(): Flow<Long> = flow {
     forEach { value ->
         emit(value)

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

@@ -157,7 +157,6 @@ import kotlin.coroutines.*
  * Flow is [Reactive Streams](http://www.reactive-streams.org/) compliant, you can safely interop it with
  * reactive streams using [Flow.asPublisher] and [Publisher.asFlow] from `kotlinx-coroutines-reactive` module.
  */
-@ExperimentalCoroutinesApi
 public interface Flow<out T> {
     /**
      * Accepts the given [collector] and [emits][FlowCollector.emit] values into it.
@@ -195,7 +194,6 @@ public interface Flow<out T> {
  * }
  * ```
  */
-@FlowPreview
 public abstract class AbstractFlow<T> : Flow<T> {
 
     @InternalCoroutinesApi

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

@@ -4,16 +4,13 @@
 
 package kotlinx.coroutines.flow
 
-import kotlinx.coroutines.*
-
 /**
  * [FlowCollector] is used as an intermediate or a terminal collector of the flow and represents
  * an entity that accepts values emitted by the [Flow].
  *
  * This interface should usually not be implemented directly, but rather used as a receiver in a [flow] builder when implementing a custom operator.
  * Implementations of this interface are not thread-safe.
  */
-@ExperimentalCoroutinesApi
 public interface FlowCollector<in T> {
 
     /**

kotlinx-coroutines-core/common/src/flow/internal/NopCollector.kt

@@ -4,8 +4,6 @@
 
 package kotlinx.coroutines.flow.internal
 
-import kotlinx.coroutines.flow.*
-
 internal object NopCollector : ConcurrentFlowCollector<Any?> {
     override suspend fun emit(value: Any?) {
         // does nothing

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

@@ -265,4 +265,3 @@ private fun checkFlowContext(context: CoroutineContext) {
         "Flow context cannot contain job in it. Had $context"
     }
 }
-

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

@@ -17,46 +17,40 @@ import kotlinx.coroutines.flow.unsafeTransform as transform
 /**
  * Returns a flow containing only values of the original flow that matches the given [predicate].
  */
-@ExperimentalCoroutinesApi
 public inline fun <T> Flow<T>.filter(crossinline predicate: suspend (T) -> Boolean): Flow<T> = transform { value ->
     if (predicate(value)) return@transform emit(value)
 }
 
 /**
  * Returns a flow containing only values of the original flow that do not match the given [predicate].
  */
-@ExperimentalCoroutinesApi
 public inline fun <T> Flow<T>.filterNot(crossinline predicate: suspend (T) -> Boolean): Flow<T> = transform { value ->
     if (!predicate(value)) return@transform emit(value)
 }
 
 /**
  * Returns a flow containing only values that are instances of specified type [R].
  */
-@ExperimentalCoroutinesApi
 @Suppress("UNCHECKED_CAST")
 public inline fun <reified R> Flow<*>.filterIsInstance(): Flow<R> = filter { it is R } as Flow<R>
 
 /**
  * Returns a flow containing only values of the original flow that are not null.
  */
-@ExperimentalCoroutinesApi
 public fun <T: Any> Flow<T?>.filterNotNull(): Flow<T> = transform<T?, T> { value ->
     if (value != null) return@transform emit(value)
 }
 
 /**
  * Returns a flow containing the results of applying the given [transform] function to each value of the original flow.
  */
-@ExperimentalCoroutinesApi
 public inline fun <T, R> Flow<T>.map(crossinline transform: suspend (value: T) -> R): Flow<R> = transform { value ->
    return@transform emit(transform(value))
 }
 
 /**
  * Returns a flow that contains only non-null results of applying the given [transform] function to each value of the original flow.
  */
-@ExperimentalCoroutinesApi
 public inline fun <T, R: Any> Flow<T>.mapNotNull(crossinline transform: suspend (value: T) -> R?): Flow<R> = transform { value ->
     val transformed = transform(value) ?: return@transform
     return@transform emit(transformed)
@@ -65,7 +59,6 @@ public inline fun <T, R: Any> Flow<T>.mapNotNull(crossinline transform: suspend
 /**
  * Returns a flow which performs the given [action] on each value of the original flow.
  */
-@ExperimentalCoroutinesApi
 public fun <T> Flow<T>.onEach(action: suspend (T) -> Unit): Flow<T> = transform { value ->
     action(value)
     return@transform emit(value)

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

@@ -27,7 +27,6 @@ import kotlin.jvm.*
  *     .collect() // trigger collection of the flow
  * ```
  */
-@ExperimentalCoroutinesApi // tentatively stable in 1.3.0
 public suspend fun Flow<*>.collect() = collect(NopCollector)
 
 /**
@@ -69,7 +68,6 @@ public fun <T> Flow<T>.launchIn(scope: CoroutineScope): Job = scope.launch {
  * }
  * ```
  */
-@ExperimentalCoroutinesApi
 public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value: T) -> Unit): Unit =
     collect(object : FlowCollector<T> {
         override suspend fun emit(value: T) = action(value)

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

@@ -7,25 +7,21 @@
 
 package kotlinx.coroutines.flow
 
-import kotlinx.coroutines.*
 import kotlin.jvm.*
 
 /**
  * Collects given flow into a [destination]
  */
-@ExperimentalCoroutinesApi
 public suspend fun <T> Flow<T>.toList(destination: MutableList<T> = ArrayList()): List<T> = toCollection(destination)
 
 /**
  * Collects given flow into a [destination]
  */
-@ExperimentalCoroutinesApi
 public suspend fun <T> Flow<T>.toSet(destination: MutableSet<T> = LinkedHashSet()): Set<T> = toCollection(destination)
 
 /**
  * Collects given flow into a [destination]
  */
-@ExperimentalCoroutinesApi
 public suspend fun <T, C : MutableCollection<in T>> Flow<T>.toCollection(destination: C): C {
     collect { value ->
         destination.add(value)

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

@@ -27,7 +27,7 @@ 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  {
+public suspend fun <T> Flow<T>.count(predicate: suspend (T) -> Boolean): Int {
     var i = 0
     collect { value ->
         if (predicate(value)) {

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

@@ -55,7 +55,6 @@ public suspend inline fun <T, R> Flow<T>.fold(
  * Throws [NoSuchElementException] for empty flow and [IllegalStateException] for flow
  * that contains more than one element.
  */
-@ExperimentalCoroutinesApi
 public suspend fun <T> Flow<T>.single(): T {
     var result: Any? = NULL
     collect { value ->
@@ -72,7 +71,6 @@ public suspend fun <T> Flow<T>.single(): T {
  * The terminal operator, that awaits for one and only one value to be published.
  * Throws [IllegalStateException] for flow that contains more than one element.
  */
-@ExperimentalCoroutinesApi
 public suspend fun <T: Any> Flow<T>.singleOrNull(): T? {
     var result: T? = null
     collect { value ->
@@ -87,7 +85,6 @@ public suspend fun <T: Any> Flow<T>.singleOrNull(): T? {
  * The terminal operator that returns the first element emitted by the flow and then cancels flow's collection.
  * Throws [NoSuchElementException] if the flow was empty.
  */
-@ExperimentalCoroutinesApi
 public suspend fun <T> Flow<T>.first(): T {
     var result: Any? = NULL
     try {
@@ -107,7 +104,6 @@ public suspend fun <T> Flow<T>.first(): T {
  * The terminal operator that returns the first element emitted by the flow matching the given [predicate] and then cancels flow's collection.
  * Throws [NoSuchElementException] if the flow has not contained elements matching the [predicate].
  */
-@ExperimentalCoroutinesApi
 public suspend fun <T> Flow<T>.first(predicate: suspend (T) -> Boolean): T {
     var result: Any? = NULL
     try {