From 328eda32743d094699d66244cfdced1788bafb64 Mon Sep 17 00:00:00 2001 From: Artem Kobzar Date: Fri, 3 Jul 2026 20:25:25 +0200 Subject: [PATCH 1/5] [K/JS] Add exportability of Flow with a conversion method to `AsyncIterator` With the following commit, the `Flow` interface could be consumed on the JavaScript / TypeScript side as follows: ```typescript for await (const value of FLOW_PRODUCER.asAsyncIterable()) { ... } ``` As well as created from either `AsyncIterable`, `AsyncIterator` or an async generator. With an async iterator the API looks like the following: ```typescript import { Flow } from "coroutines" Flow.fromAsyncGenerator(async function* () { yield await fetch(...) yield "Hello, World" }); ``` ^KT-80733 Fixed --- .../common/src/flow/Flow.kt | 2 +- .../concurrent/src/flow/Flow.concurrent.kt | 195 ++++++++ .../js/src/channels/Channel.kt | 28 +- .../js/src/flow/Flow.js.kt | 268 ++++++++++ .../src/internal/JsAsyncIteratorProtocol.kt | 26 + .../js/test/ChannelInteropTest.kt | 1 + .../js/test/FlowInteropTest.kt | 460 ++++++++++++++++++ .../wasmJs/src/flow/Flow.wasm.kt | 195 ++++++++ .../wasmWasi/src/flow/Flow.wasm.kt | 195 ++++++++ 9 files changed, 1345 insertions(+), 25 deletions(-) create mode 100644 kotlinx-coroutines-core/concurrent/src/flow/Flow.concurrent.kt create mode 100644 kotlinx-coroutines-core/js/src/flow/Flow.js.kt create mode 100644 kotlinx-coroutines-core/js/src/internal/JsAsyncIteratorProtocol.kt create mode 100644 kotlinx-coroutines-core/js/test/FlowInteropTest.kt create mode 100644 kotlinx-coroutines-core/wasmJs/src/flow/Flow.wasm.kt create mode 100644 kotlinx-coroutines-core/wasmWasi/src/flow/Flow.wasm.kt diff --git a/kotlinx-coroutines-core/common/src/flow/Flow.kt b/kotlinx-coroutines-core/common/src/flow/Flow.kt index 5735028847..f09e7a85b4 100644 --- a/kotlinx-coroutines-core/common/src/flow/Flow.kt +++ b/kotlinx-coroutines-core/common/src/flow/Flow.kt @@ -173,7 +173,7 @@ import kotlin.coroutines.* * These implementations ensure that the context preservation property is not violated, and prevent most * of the developer mistakes related to concurrency, inconsistent flow dispatchers, and cancellation. */ -public interface Flow { +public expect interface Flow { /** * Accepts the given [collector] and [emits][FlowCollector.emit] values into it. diff --git a/kotlinx-coroutines-core/concurrent/src/flow/Flow.concurrent.kt b/kotlinx-coroutines-core/concurrent/src/flow/Flow.concurrent.kt new file mode 100644 index 0000000000..a8c3ccb92f --- /dev/null +++ b/kotlinx-coroutines-core/concurrent/src/flow/Flow.concurrent.kt @@ -0,0 +1,195 @@ +package kotlinx.coroutines.flow + +import kotlinx.coroutines.* +import kotlinx.coroutines.flow.internal.* +import kotlin.coroutines.* + +/** + * An asynchronous data stream that sequentially emits values and completes normally or with an exception. + * + * _Intermediate operators_ on the flow such as [map], [filter], [take], [zip], etc are functions that are + * applied to the _upstream_ flow or flows and return a _downstream_ flow where further operators can be applied to. + * Intermediate operations do not execute any code in the flow and are not suspending functions themselves. + * They only set up a chain of operations for future execution and quickly return. + * This is known as a _cold flow_ property. + * + * _Terminal operators_ on the flow are either suspending functions such as [collect], [single], [reduce], [toList], etc. + * or [launchIn] operator that starts collection of the flow in the given scope. + * They are applied to the upstream flow and trigger execution of all operations. + * Execution of the flow is also called _collecting the flow_ and is always performed in a suspending manner + * without actual blocking. Terminal operators complete normally or exceptionally depending on successful or failed + * execution of all the flow operations in the upstream. The most basic terminal operator is [collect], for example: + * + * ``` + * try { + * flow.collect { value -> + * println("Received $value") + * } + * } catch (e: Exception) { + * println("The flow has thrown an exception: $e") + * } + * ``` + * + * By default, flows are _sequential_ and all flow operations are executed sequentially in the same coroutine, + * with an exception for a few operations specifically designed to introduce concurrency into flow + * execution such as [buffer] and [flatMapMerge]. See their documentation for details. + * + * The `Flow` interface does not carry information whether a flow is a _cold_ stream that can be collected repeatedly and + * triggers execution of the same code every time it is collected, or if it is a _hot_ stream that emits different + * values from the same running source on each collection. Usually flows represent _cold_ streams, but + * there is a [SharedFlow] subtype that represents _hot_ streams. In addition to that, any flow can be turned + * into a _hot_ one by the [stateIn] and [shareIn] operators, or by converting the flow into a hot channel + * via the [produceIn] operator. + * + * ### Flow builders + * + * There are the following basic ways to create a flow: + * + * - [flowOf(...)][flowOf] functions to create a flow from a fixed set of values. + * - [asFlow()][asFlow] extension functions on various types to convert them into flows. + * - [flow { ... }][flow] builder function to construct arbitrary flows from + * sequential calls to [emit][FlowCollector.emit] function. + * - [channelFlow { ... }][channelFlow] builder function to construct arbitrary flows from + * potentially concurrent calls to the [send][kotlinx.coroutines.channels.SendChannel.send] function. + * - [MutableStateFlow] and [MutableSharedFlow] define the corresponding constructor functions to create + * a _hot_ flow that can be directly updated. + * + * ### Flow constraints + * + * All implementations of the `Flow` interface must adhere to two key properties described in detail below: + * + * - Context preservation. + * - Exception transparency. + * + * These properties ensure the ability to perform local reasoning about the code with flows and modularize the code + * in such a way that upstream flow emitters can be developed separately from downstream flow collectors. + * A user of a flow does not need to be aware of implementation details of the upstream flows it uses. + * + * ### Context preservation + * + * The flow has a context preservation property: it encapsulates its own execution context and never propagates or leaks + * it downstream, thus making reasoning about the execution context of particular transformations or terminal + * operations trivial. + * + * There is only one way to change the context of a flow: the [flowOn][Flow.flowOn] operator + * that changes the upstream context ("everything above the `flowOn` operator"). + * For additional information refer to its documentation. + * + * This reasoning can be demonstrated in practice: + * + * ``` + * val flowA = flowOf(1, 2, 3) + * .map { it + 1 } // Will be executed in ctxA + * .flowOn(ctxA) // Changes the upstream context: flowOf and map + * + * // Now we have a context-preserving flow: it is executed somewhere but this information is encapsulated in the flow itself + * + * val filtered = flowA // ctxA is encapsulated in flowA + * .filter { it == 3 } // Pure operator without a context yet + * + * withContext(Dispatchers.Main) { + * // All non-encapsulated operators will be executed in Main: filter and single + * val result = filtered.single() + * myUi.text = result + * } + * ``` + * + * From the implementation point of view, it means that all flow implementations should + * only emit from the same coroutine context. + * This constraint is efficiently enforced by the default [flow] builder. + * The [flow] builder should be used if the flow implementation does not start any coroutines. + * Its implementation prevents most of the development mistakes: + * + * ``` + * val myFlow = flow { + * // GlobalScope.launch { // is prohibited + * // launch(Dispatchers.IO) { // is prohibited + * // withContext(CoroutineName("myFlow")) { // is prohibited + * emit(1) // OK + * coroutineScope { + * emit(2) // OK -- still the same coroutine + * } + * } + * ``` + * + * Use [channelFlow] if the collection and emission of a flow are to be separated into multiple coroutines. + * It encapsulates all the context preservation work and allows you to focus on your + * domain-specific problem, rather than invariant implementation details. + * It is possible to use any combination of coroutine builders from within [channelFlow]. + * + * If you are looking for performance and are sure that no concurrent emits and context jumps will happen, + * the [flow] builder can be used alongside a [coroutineScope] or [supervisorScope] instead: + * - Scoped primitive should be used to provide a [CoroutineScope]. + * - Changing the context of emission is prohibited, no matter whether it is `withContext(ctx)` or + * a builder argument (e.g. `launch(ctx)`). + * - Collecting another flow from a separate context is allowed, but it has the same effect as + * applying the [flowOn] operator to that flow, which is more efficient. + * + * ### Exception transparency + * + * When `emit` or `emitAll` throws, the Flow implementations must immediately stop emitting new values and finish with an exception. + * For diagnostics or application-specific purposes, the exception may be different from the one thrown by the emit operation, + * suppressing the original exception as discussed below. + * If there is a need to emit values after the downstream failed, please use the [catch][Flow.catch] operator. + * + * The [catch][Flow.catch] operator only catches upstream exceptions, but passes + * all downstream exceptions. Similarly, terminal operators like [collect][Flow.collect] + * throw any unhandled exceptions that occur in their code or in upstream flows, for example: + * + * ``` + * flow { emitData() } + * .map { computeOne(it) } + * .catch { ... } // catches exceptions in emitData and computeOne + * .map { computeTwo(it) } + * .collect { process(it) } // throws exceptions from process and computeTwo + * ``` + * The same reasoning can be applied to the [onCompletion] operator that is a declarative replacement for the `finally` block. + * + * All exception-handling Flow operators follow the principle of exception suppression: + * + * If the upstream flow throws an exception during its completion when the downstream exception has been thrown, + * the downstream exception becomes superseded and suppressed by the upstream exception, being a semantic + * equivalent of throwing from `finally` block. However, this doesn't affect the operation of the exception-handling operators, + * which consider the downstream exception to be the root cause and behave as if the upstream didn't throw anything. + * + * Failure to adhere to the exception transparency requirement can lead to strange behaviors which make + * it hard to reason about the code because an exception in the `collect { ... }` could be somehow "caught" + * by an upstream flow, limiting the ability of local reasoning about the code. + * + * Flow machinery enforces exception transparency at runtime and throws [IllegalStateException] on any attempt to emit a value, + * if an exception has been thrown on previous attempt. + * + * ### Reactive streams + * + * 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. + * + * ### Not stable for inheritance + * + * **The `Flow` interface is not stable for inheritance in 3rd party libraries**, as new methods + * might be added to this interface in the future, but is stable for use. + * + * Use the `flow { ... }` builder function to create an implementation, or extend [AbstractFlow]. + * These implementations ensure that the context preservation property is not violated, and prevent most + * of the developer mistakes related to concurrency, inconsistent flow dispatchers, and cancellation. + */ +public actual interface Flow { + + /** + * Accepts the given [collector] and [emits][FlowCollector.emit] values into it. + * + * This method can be used along with SAM-conversion of [FlowCollector]: + * ``` + * myFlow.collect { value -> println("Collected $value") } + * ``` + * + * ### Method inheritance + * + * To ensure the context preservation property, it is not recommended implementing this method directly. + * Instead, [AbstractFlow] can be used as the base type to properly ensure flow's properties. + * + * All default flow implementations ensure context preservation and exception transparency properties on a best-effort basis + * and throw [IllegalStateException] if a violation was detected. + */ + public actual suspend fun collect(collector: FlowCollector) +} diff --git a/kotlinx-coroutines-core/js/src/channels/Channel.kt b/kotlinx-coroutines-core/js/src/channels/Channel.kt index f2a7945775..731be613bf 100644 --- a/kotlinx-coroutines-core/js/src/channels/Channel.kt +++ b/kotlinx-coroutines-core/js/src/channels/Channel.kt @@ -1,13 +1,15 @@ -@file:OptIn(ExperimentalJsExport::class, ExperimentalStdlibApi::class) +@file:OptIn(ExperimentalJsExport::class) @file:Suppress("EXPOSED_FUNCTION_RETURN_TYPE", "INVISIBLE_REFERENCE", "EXPOSED_SUPER_INTERFACE") package kotlinx.coroutines.channels import kotlinx.coroutines.* +import kotlinx.coroutines.internal.JsAsyncIterable import kotlinx.coroutines.internal.recoverStackTrace import kotlinx.coroutines.selects.* -import kotlinx.js.JsPlainObject import kotlin.internal.* import kotlin.js.Promise +import kotlinx.coroutines.internal.JsAsyncIterator +import kotlinx.coroutines.internal.JsIteratorResult @JsImplicitExport(couldBeConvertedToExplicitExport = true) public actual interface ReceiveChannel : JsAsyncIterable { @@ -132,25 +134,3 @@ public actual interface ReceiveChannel : JsAsyncIterable { public actual val onReceiveOrNull: SelectClause1 get() = (this as BufferedChannel).onReceiveOrNull } -@JsName("AsyncIterable") -internal external interface JsAsyncIterable { - @JsSymbol("asyncIterator") - public fun asyncIterator(): JsAsyncIterator -} - -@JsPlainObject -@JsName("AsyncIterator") -// https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#the_async_iterator_and_async_iterable_protocols -internal external interface JsAsyncIterator { - public val next: () -> Promise> - // `return` and `throw` must be able to accept either zero arguments or a single one - public val `return`: (value: @UnsafeVariance T?) -> Promise> - public val `throw`: (value: Any?) -> Promise> -} - -@JsPlainObject -@JsName("IteratorResult") -internal external interface JsIteratorResult { - public val value: T? - public val done: Boolean -} diff --git a/kotlinx-coroutines-core/js/src/flow/Flow.js.kt b/kotlinx-coroutines-core/js/src/flow/Flow.js.kt new file mode 100644 index 0000000000..741a51d8ed --- /dev/null +++ b/kotlinx-coroutines-core/js/src/flow/Flow.js.kt @@ -0,0 +1,268 @@ +@file:OptIn(ExperimentalJsExport::class, ExperimentalJsStatic::class, ExperimentalStdlibApi::class) +@file:Suppress("INVISIBLE_REFERENCE", "EXPOSED_FUNCTION_RETURN_TYPE", "EXPOSED_PARAMETER_TYPE") +package kotlinx.coroutines.flow + +import kotlinx.coroutines.* +import kotlinx.coroutines.internal.JsAsyncIterable +import kotlinx.coroutines.internal.JsAsyncIterator +import kotlin.coroutines.EmptyCoroutineContext + +/** + * An asynchronous data stream that sequentially emits values and completes normally or with an exception. + * + * _Intermediate operators_ on the flow such as [map], [filter], [take], [zip], etc are functions that are + * applied to the _upstream_ flow or flows and return a _downstream_ flow where further operators can be applied to. + * Intermediate operations do not execute any code in the flow and are not suspending functions themselves. + * They only set up a chain of operations for future execution and quickly return. + * This is known as a _cold flow_ property. + * + * _Terminal operators_ on the flow are either suspending functions such as [collect], [single], [reduce], [toList], etc. + * or [launchIn] operator that starts collection of the flow in the given scope. + * They are applied to the upstream flow and trigger execution of all operations. + * Execution of the flow is also called _collecting the flow_ and is always performed in a suspending manner + * without actual blocking. Terminal operators complete normally or exceptionally depending on successful or failed + * execution of all the flow operations in the upstream. The most basic terminal operator is [collect], for example: + * + * ``` + * try { + * flow.collect { value -> + * println("Received $value") + * } + * } catch (e: Exception) { + * println("The flow has thrown an exception: $e") + * } + * ``` + * + * By default, flows are _sequential_ and all flow operations are executed sequentially in the same coroutine, + * with an exception for a few operations specifically designed to introduce concurrency into flow + * execution such as [buffer] and [flatMapMerge]. See their documentation for details. + * + * The `Flow` interface does not carry information whether a flow is a _cold_ stream that can be collected repeatedly and + * triggers execution of the same code every time it is collected, or if it is a _hot_ stream that emits different + * values from the same running source on each collection. Usually flows represent _cold_ streams, but + * there is a [SharedFlow] subtype that represents _hot_ streams. In addition to that, any flow can be turned + * into a _hot_ one by the [stateIn] and [shareIn] operators, or by converting the flow into a hot channel + * via the [produceIn] operator. + * + * ### Flow builders + * + * There are the following basic ways to create a flow: + * + * - [flowOf(...)][flowOf] functions to create a flow from a fixed set of values. + * - [asFlow()][asFlow] extension functions on various types to convert them into flows. + * - [flow { ... }][flow] builder function to construct arbitrary flows from + * sequential calls to [emit][FlowCollector.emit] function. + * - [channelFlow { ... }][channelFlow] builder function to construct arbitrary flows from + * potentially concurrent calls to the [send][kotlinx.coroutines.channels.SendChannel.send] function. + * - [MutableStateFlow] and [MutableSharedFlow] define the corresponding constructor functions to create + * a _hot_ flow that can be directly updated. + * + * ### Flow constraints + * + * All implementations of the `Flow` interface must adhere to two key properties described in detail below: + * + * - Context preservation. + * - Exception transparency. + * + * These properties ensure the ability to perform local reasoning about the code with flows and modularize the code + * in such a way that upstream flow emitters can be developed separately from downstream flow collectors. + * A user of a flow does not need to be aware of implementation details of the upstream flows it uses. + * + * ### Context preservation + * + * The flow has a context preservation property: it encapsulates its own execution context and never propagates or leaks + * it downstream, thus making reasoning about the execution context of particular transformations or terminal + * operations trivial. + * + * There is only one way to change the context of a flow: the [flowOn][Flow.flowOn] operator + * that changes the upstream context ("everything above the `flowOn` operator"). + * For additional information refer to its documentation. + * + * This reasoning can be demonstrated in practice: + * + * ``` + * val flowA = flowOf(1, 2, 3) + * .map { it + 1 } // Will be executed in ctxA + * .flowOn(ctxA) // Changes the upstream context: flowOf and map + * + * // Now we have a context-preserving flow: it is executed somewhere but this information is encapsulated in the flow itself + * + * val filtered = flowA // ctxA is encapsulated in flowA + * .filter { it == 3 } // Pure operator without a context yet + * + * withContext(Dispatchers.Main) { + * // All non-encapsulated operators will be executed in Main: filter and single + * val result = filtered.single() + * myUi.text = result + * } + * ``` + * + * From the implementation point of view, it means that all flow implementations should + * only emit from the same coroutine context. + * This constraint is efficiently enforced by the default [flow] builder. + * The [flow] builder should be used if the flow implementation does not start any coroutines. + * Its implementation prevents most of the development mistakes: + * + * ``` + * val myFlow = flow { + * // GlobalScope.launch { // is prohibited + * // launch(Dispatchers.IO) { // is prohibited + * // withContext(CoroutineName("myFlow")) { // is prohibited + * emit(1) // OK + * coroutineScope { + * emit(2) // OK -- still the same coroutine + * } + * } + * ``` + * + * Use [channelFlow] if the collection and emission of a flow are to be separated into multiple coroutines. + * It encapsulates all the context preservation work and allows you to focus on your + * domain-specific problem, rather than invariant implementation details. + * It is possible to use any combination of coroutine builders from within [channelFlow]. + * + * If you are looking for performance and are sure that no concurrent emits and context jumps will happen, + * the [flow] builder can be used alongside a [coroutineScope] or [supervisorScope] instead: + * - Scoped primitive should be used to provide a [CoroutineScope]. + * - Changing the context of emission is prohibited, no matter whether it is `withContext(ctx)` or + * a builder argument (e.g. `launch(ctx)`). + * - Collecting another flow from a separate context is allowed, but it has the same effect as + * applying the [flowOn] operator to that flow, which is more efficient. + * + * ### Exception transparency + * + * When `emit` or `emitAll` throws, the Flow implementations must immediately stop emitting new values and finish with an exception. + * For diagnostics or application-specific purposes, the exception may be different from the one thrown by the emit operation, + * suppressing the original exception as discussed below. + * If there is a need to emit values after the downstream failed, please use the [catch][Flow.catch] operator. + * + * The [catch][Flow.catch] operator only catches upstream exceptions, but passes + * all downstream exceptions. Similarly, terminal operators like [collect][Flow.collect] + * throw any unhandled exceptions that occur in their code or in upstream flows, for example: + * + * ``` + * flow { emitData() } + * .map { computeOne(it) } + * .catch { ... } // catches exceptions in emitData and computeOne + * .map { computeTwo(it) } + * .collect { process(it) } // throws exceptions from process and computeTwo + * ``` + * The same reasoning can be applied to the [onCompletion] operator that is a declarative replacement for the `finally` block. + * + * All exception-handling Flow operators follow the principle of exception suppression: + * + * If the upstream flow throws an exception during its completion when the downstream exception has been thrown, + * the downstream exception becomes superseded and suppressed by the upstream exception, being a semantic + * equivalent of throwing from `finally` block. However, this doesn't affect the operation of the exception-handling operators, + * which consider the downstream exception to be the root cause and behave as if the upstream didn't throw anything. + * + * Failure to adhere to the exception transparency requirement can lead to strange behaviors which make + * it hard to reason about the code because an exception in the `collect { ... }` could be somehow "caught" + * by an upstream flow, limiting the ability of local reasoning about the code. + * + * Flow machinery enforces exception transparency at runtime and throws [IllegalStateException] on any attempt to emit a value, + * if an exception has been thrown on previous attempt. + * + * ### Reactive streams + * + * 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. + * + * ### Not stable for inheritance + * + * **The `Flow` interface is not stable for inheritance in 3rd party libraries**, as new methods + * might be added to this interface in the future, but is stable for use. + * + * Use the `flow { ... }` builder function to create an implementation, or extend [AbstractFlow]. + * These implementations ensure that the context preservation property is not violated, and prevent most + * of the developer mistakes related to concurrency, inconsistent flow dispatchers, and cancellation. + */ +public actual interface Flow { + + /** + * Accepts the given [collector] and [emits][FlowCollector.emit] values into it. + * + * This method can be used along with SAM-conversion of [FlowCollector]: + * ``` + * myFlow.collect { value -> println("Collected $value") } + * ``` + * + * ### Method inheritance + * + * To ensure the context preservation property, it is not recommended implementing this method directly. + * Instead, [AbstractFlow] can be used as the base type to properly ensure flow's properties. + * + * All default flow implementations ensure context preservation and exception transparency properties on a best-effort basis + * and throw [IllegalStateException] if a violation was detected. + */ + @JsExport.Ignore + public actual suspend fun collect(collector: FlowCollector) + + + + @JsExport.Ignore + // For Kotlin side only to be able to set up a custom scope for the iterator + public fun asAsyncIterable(scope: CoroutineScope): JsAsyncIterable = + buffer(0).produceIn(scope) + + public fun asAsyncIterable(): JsAsyncIterable = + asAsyncIterable(CoroutineScope(EmptyCoroutineContext)) + + @JsExport.Ignore + // Important note: it would be much nicer to place those factory functions outside of Flow + // so from both Kotlin and TypeScript side it could be used without importing Flow (like in `flowOf` or `flow`) + // However, the described way of exporting factory functions forces the functions always to be exported (even if people don't use them and don't export Flow), + // and that may cause bundle size problems (at least right now). + // So, until the bundle size problem is solved, we keep those factory functions inside Flow, with possibility to move them outside later. + public companion object { + /** + * Converts a JavaScript AsyncIterable to a Kotlin Flow. + * + * The resulting flow will iterate through all values produced by the async iterable. + * If the flow collection is canceled or fails, the iterator's `return()` method will be called + * to properly clean up the async iterable. + */ + @JsStatic + public fun from(async: JsAsyncIterable): Flow = + from(async.asyncIterator()) + + /** + * Converts a JavaScript async generator function to a Kotlin Flow. + * + * The generator will be invoked to get an async iterator for collection. + * Cancellation or failure during a collection triggers the iterator's `return()` method + * to ensure proper cleanup. + */ + @JsStatic + @JsName("fromAsyncGenerator") + public fun from(generator: () -> JsAsyncIterator): Flow = flow { + var completed = false + val iterator = generator() + try { + while (true) { + val result = iterator.next().await() + if (result.done) { + completed = true + break + } + emit(result.value.unsafeCast()) + } + } finally { + if (!completed) { + iterator.`return`().await() + } + } + } + + /** + * Converts a JavaScript AsyncIterator to a Kotlin Flow. + * + * The resulting flow emits items produced by the iterator until it reports completion. + * If a collection is canceled or fails, the iterator's `return()` method is called + * to close the iterator. + */ + @JsStatic + @JsName("fromAsyncIterator") + public fun from(iterator: JsAsyncIterator): Flow = + from { iterator } + } +} diff --git a/kotlinx-coroutines-core/js/src/internal/JsAsyncIteratorProtocol.kt b/kotlinx-coroutines-core/js/src/internal/JsAsyncIteratorProtocol.kt new file mode 100644 index 0000000000..843375cf13 --- /dev/null +++ b/kotlinx-coroutines-core/js/src/internal/JsAsyncIteratorProtocol.kt @@ -0,0 +1,26 @@ +@file:OptIn(ExperimentalStdlibApi::class) +package kotlinx.coroutines.internal + +import kotlinx.js.JsPlainObject +import kotlin.js.Promise + +@JsPlainObject +@JsName("AsyncIterator") +internal external interface JsAsyncIterator { + public val next: () -> Promise> + public val `return`: (value: @UnsafeVariance T?) -> Promise> + public val `throw`: (value: Any?) -> Promise> +} + +@JsPlainObject +@JsName("IteratorResult") +internal external interface JsIteratorResult { + public val value: T? + public val done: Boolean +} + +@JsName("AsyncIterable") +internal external interface JsAsyncIterable { + @JsSymbol("asyncIterator") + public fun asyncIterator(): JsAsyncIterator +} diff --git a/kotlinx-coroutines-core/js/test/ChannelInteropTest.kt b/kotlinx-coroutines-core/js/test/ChannelInteropTest.kt index 49133fe309..380ff0b964 100644 --- a/kotlinx-coroutines-core/js/test/ChannelInteropTest.kt +++ b/kotlinx-coroutines-core/js/test/ChannelInteropTest.kt @@ -1,6 +1,7 @@ package kotlinx.coroutines import kotlinx.coroutines.channels.* +import kotlinx.coroutines.internal.JsAsyncIterator import kotlinx.coroutines.testing.* import kotlin.js.* import kotlin.test.* diff --git a/kotlinx-coroutines-core/js/test/FlowInteropTest.kt b/kotlinx-coroutines-core/js/test/FlowInteropTest.kt new file mode 100644 index 0000000000..dcf6f5274f --- /dev/null +++ b/kotlinx-coroutines-core/js/test/FlowInteropTest.kt @@ -0,0 +1,460 @@ +package kotlinx.coroutines + +import kotlinx.coroutines.flow.* +import kotlinx.coroutines.internal.JsAsyncIterable +import kotlinx.coroutines.testing.* +import kotlin.js.* +import kotlin.test.* +import kotlinx.coroutines.internal.JsAsyncIterator +import kotlinx.coroutines.internal.JsIteratorResult + +class FlowInteropTest : TestBase() { + + // ===== Flow to AsyncIterator tests ===== + + @Test + fun testFlowToAsyncIteratorBasic() = runTest { + val flow = flowOf(1, 2, 3) + val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() + + val result1 = iterator.next().await() + assertEquals(1, result1.value) + assertEquals(false, result1.done) + + val result2 = iterator.next().await() + assertEquals(2, result2.value) + assertEquals(false, result2.done) + + val result3 = iterator.next().await() + assertEquals(3, result3.value) + assertEquals(false, result3.done) + + val result4 = iterator.next().await() + assertEquals(true, result4.done) + } + + @Test + fun testFlowToAsyncIteratorEmpty() = runTest { + val flow = emptyFlow() + val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() + + val result = iterator.next().await() + assertEquals(true, result.done) + } + + @Test + fun testFlowToAsyncIteratorSingle() = runTest { + val flow = flowOf(42) + val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() + + val result1 = iterator.next().await() + assertEquals(42, result1.value) + assertEquals(false, result1.done) + + val result2 = iterator.next().await() + assertEquals(true, result2.done) + } + + @Test + fun testFlowToAsyncIteratorEarlyReturn() = runTest { + val flow = flow { + emit(1) + emit(2) + emit(3) + emit(4) + emit(5) + } + val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() + + val result1 = iterator.next().await() + assertEquals(1, result1.value) + + // Call return() to stop iteration early + val returnResult = iterator.`return`().await() + assertEquals(true, returnResult.done) + + // Next calls should return done + val result2 = iterator.next().await() + assertEquals(true, result2.done) + } + + @Test + fun testFlowToAsyncIteratorThrow() = runTest { + val flow = flow { + emit(1) + emit(2) + emit(3) + } + val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() + + val result1 = iterator.next().await() + assertEquals(1, result1.value) + + // Call throw() to cancel the iterator + val error = js("new Error('test error')") + try { + iterator.`throw`(error).await() + fail("Should have thrown") + } catch (e: Throwable) { + // Expected + } + } + + @Test + fun testFlowToAsyncIteratorException() = runTest { + val flow = flow { + emit(1) + emit(2) + throw TestException("test exception") + } + val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() + + val result1 = iterator.next().await() + assertEquals(1, result1.value) + + val result2 = iterator.next().await() + assertEquals(2, result2.value) + + // Next call should throw the exception + try { + val result = iterator.next().await() + fail("Should have thrown TestException, but return result ${result.value}") + } catch (e: TestException) { + assertEquals("test exception", e.message) + } + } + + @Test + fun testFlowToAsyncIteratorWithTransformations() = runTest { + val flow = flowOf(1, 2, 3) + .map { it * 2 } + .filter { it > 2 } + + val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() + + val result1 = iterator.next().await() + assertEquals(4, result1.value) + + val result2 = iterator.next().await() + assertEquals(6, result2.value) + + val result3 = iterator.next().await() + assertEquals(true, result3.done) + } + + @Test + fun testFlowToAsyncIteratorCancellationReturnsDone() = runTest { + val flow = flow { + emit(1) + emit(2) + emit(3) + emit(4) + emit(5) + } + val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() + + val result1 = iterator.next().await() + assertEquals(1, result1.value) + assertEquals(false, result1.done) + + val result2 = iterator.next().await() + assertEquals(2, result2.value) + assertEquals(false, result2.done) + + val returnResult = iterator.`return`().await() + assertEquals(true, returnResult.done) + + val result3 = iterator.next().await() + assertEquals(true, result3.done) + + val result4 = iterator.next().await() + assertEquals(true, result4.done) + } + + @Test + fun testFlowToAsyncIteratorCancellationExceptionReturnsDone() = runTest { + val flow = flow { + emit(1) + emit(2) + emit(3) + throw CancellationException("flow cancelled") + emit(4) // These should never be emitted + emit(5) + emit(6) + } + val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() + + val result1 = iterator.next().await() + assertEquals(1, result1.value) + assertEquals(false, result1.done) + + val result2 = iterator.next().await() + assertEquals(2, result2.value) + assertEquals(false, result2.done) + + val result3 = iterator.next().await() + assertEquals(3, result3.value) + assertEquals(false, result3.done) + + // When flow throws CancellationException, next() should return done: true + // without propagating the exception, and unconditionally stop emitting + val result4 = iterator.next().await() + assertEquals(true, result4.done) + + // Subsequent calls should also return done: true, ensuring elements after + // CancellationException are never emitted + val result5 = iterator.next().await() + assertEquals(true, result5.done) + + val result6 = iterator.next().await() + assertEquals(true, result6.done) + } + + // ===== AsyncIterator to Flow tests ===== + + @Test + fun testAsyncIteratorToFlowBasic() = runTest { + val asyncIterator = createAsyncIterator(listOf(1, 2, 3)) + val flow = Flow.from(asyncIterator) + + val results = mutableListOf() + flow.collect { results.add(it) } + + assertEquals(listOf(1, 2, 3), results) + } + + @Test + fun testAsyncIteratorToFlowEmpty() = runTest { + val asyncIterator = createAsyncIterator(emptyList()) + val flow = Flow.from(asyncIterator) + + val results = mutableListOf() + flow.collect { results.add(it) } + + assertEquals(emptyList(), results) + } + + @Test + fun testAsyncIteratorToFlowSingle() = runTest { + val asyncIterator = createAsyncIterator(listOf(42)) + val flow = Flow.from(asyncIterator) + + val results = mutableListOf() + flow.collect { results.add(it) } + + assertEquals(listOf(42), results) + } + + @Test + fun testAsyncIteratorToFlowCancellation() = runTest { + var returnCalled = false + val asyncIterator = createAsyncIteratorWithCleanup( + listOf(1, 2, 3, 4, 5), + onReturn = { returnCalled = true } + ) + val flow = Flow.from(asyncIterator) + + val results = mutableListOf() + flow.take(2).collect { results.add(it) } + + assertEquals(listOf(1, 2), results) + yield() // Allow cleanup to happen + assertTrue(returnCalled, "return() should be called on cancellation") + } + + @Test + fun testAsyncIteratorToFlowException() = runTest { + val asyncIterator = createAsyncIteratorWithException( + listOf(1, 2), + TestException("iterator error") + ) + val flow = Flow.from(asyncIterator) + + val results = mutableListOf() + try { + flow.collect { results.add(it) } + fail("Should have thrown TestException") + } catch (e: TestException) { + assertEquals("iterator error", e.message) + } + + assertEquals(listOf(1, 2), results) + } + + @Test + fun testAsyncGeneratorToFlow() = runTest { + val generator: () -> JsAsyncIterator = { + createAsyncIterator(listOf(10, 20, 30)) + } + + val flow = Flow.from(generator) + + val results = mutableListOf() + flow.collect { results.add(it) } + + assertEquals(listOf(10, 20, 30), results) + } + + @Test + fun testAsyncIterableToFlow() = runTest { + val asyncIterable = createAsyncIterable(listOf(5, 10, 15)) + val flow = Flow.from(asyncIterable) + + val results = mutableListOf() + flow.collect { results.add(it) } + + assertEquals(listOf(5, 10, 15), results) + } + + @Test + fun testAsyncIteratorToFlowWithTransformations() = runTest { + val asyncIterator = createAsyncIterator(listOf(1, 2, 3, 4, 5)) + val flow = Flow.from(asyncIterator) + .map { it * 2 } + .filter { it > 5 } + + val results = mutableListOf() + flow.collect { results.add(it) } + + assertEquals(listOf(6, 8, 10), results) + } + + // ===== Round-trip tests ===== + @Test + fun testRoundTripFlowToAsyncIterableToFlow() = runTest { + val originalFlow = flowOf(1, 2, 3, 4, 5) + val convertedFlow = Flow.from(originalFlow.asAsyncIterable()) + + val results = mutableListOf() + convertedFlow.collect { results.add(it) } + + assertEquals(listOf(1, 2, 3, 4, 5), results) + } + + @Test + fun testRoundTripFlowToAsyncIteratorToFlow() = runTest { + val originalFlow = flowOf(1, 2, 3, 4, 5) + val iterator: JsAsyncIterator = originalFlow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() + val convertedFlow = Flow.from(iterator) + + val results = mutableListOf() + convertedFlow.collect { results.add(it) } + + assertEquals(listOf(1, 2, 3, 4, 5), results) + } + + @Test + fun testRoundTripAsyncIteratorToFlowToAsyncIterator() = runTest { + val originalIterator = createAsyncIterator(listOf(10, 20, 30)) + val flow = Flow.from(originalIterator) + val convertedIterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() + + val result1 = convertedIterator.next().await() + assertEquals(10, result1.value) + + val result2 = convertedIterator.next().await() + assertEquals(20, result2.value) + + val result3 = convertedIterator.next().await() + assertEquals(30, result3.value) + + val result4 = convertedIterator.next().await() + assertEquals(true, result4.done) + } + + // ===== Helper functions ===== + + private fun createAsyncIterator(values: List): JsAsyncIterator { + var index = 0 + val iterator = js("({})") + iterator.next = fun(): Promise> { + return if (index < values.size) { + val value = values[index++] + Promise.resolve(js("({ value: value, done: false })")) + } else { + Promise.resolve(js("({ value: undefined, done: true })")) + } + } + + iterator.`return` = fun(): Promise> { + return Promise.resolve(js("({ value: undefined, done: true })")) + } + iterator.`throw` = fun(error: Throwable): Promise> { + return Promise.reject(error) + } + return iterator + } + + private fun createAsyncIteratorWithCleanup(values: List, onReturn: () -> Unit): JsAsyncIterator { + var index = 0 + val iterator = js("({})") + iterator.next = fun(): Promise> { + return if (index < values.size) { + val value = values[index++] + Promise.resolve(js("({ value: value, done: false })")) + } else { + Promise.resolve(js("({ value: undefined, done: true })")) + } + } + + iterator.`return` = fun(): Promise> { + onReturn() + return Promise.resolve(js("({ value: undefined, done: true })")) + } + iterator.`throw` = fun(error: Throwable): Promise> { + return Promise.reject(error) + } + return iterator + } + + private fun createAsyncIteratorWithException(values: List, exception: Throwable): JsAsyncIterator { + var index = 0 + val iterator = js("({})") + iterator.next = fun(): Promise> { + return if (index < values.size) { + val value = values[index++] + Promise.resolve(js("({ value: value, done: false })")) + } else { + Promise.reject(exception) + } + } + + iterator.`return` = fun(): Promise> { + return Promise.resolve(js("({ value: undefined, done: true })")) + } + iterator.`throw` = fun(error: Throwable): Promise> { + return Promise.reject(error) + } + return iterator + } + + private fun createAsyncIteratorWithCallCounter(values: List, onNextCall: () -> Unit): JsAsyncIterator { + var index = 0 + val iterator = js("({})") + iterator.next = fun(): Promise> { + onNextCall() + return if (index < values.size) { + val value = values[index++] + Promise.resolve(js("({ value: value, done: false })")) + } else { + Promise.resolve(js("({ value: undefined, done: true })")) + } + } + + iterator.`return` = fun(): Promise> { + return Promise.resolve(js("({ value: undefined, done: true })")) + } + iterator.`throw` = fun(error: Throwable): Promise> { + return Promise.reject(error) + } + return iterator + } + + private fun createAsyncIterable(values: List): JsAsyncIterable { + val iterable = js("({})") + iterable[js("Symbol.asyncIterator")] = { + createAsyncIterator(values) + } + return iterable + } +} diff --git a/kotlinx-coroutines-core/wasmJs/src/flow/Flow.wasm.kt b/kotlinx-coroutines-core/wasmJs/src/flow/Flow.wasm.kt new file mode 100644 index 0000000000..a8c3ccb92f --- /dev/null +++ b/kotlinx-coroutines-core/wasmJs/src/flow/Flow.wasm.kt @@ -0,0 +1,195 @@ +package kotlinx.coroutines.flow + +import kotlinx.coroutines.* +import kotlinx.coroutines.flow.internal.* +import kotlin.coroutines.* + +/** + * An asynchronous data stream that sequentially emits values and completes normally or with an exception. + * + * _Intermediate operators_ on the flow such as [map], [filter], [take], [zip], etc are functions that are + * applied to the _upstream_ flow or flows and return a _downstream_ flow where further operators can be applied to. + * Intermediate operations do not execute any code in the flow and are not suspending functions themselves. + * They only set up a chain of operations for future execution and quickly return. + * This is known as a _cold flow_ property. + * + * _Terminal operators_ on the flow are either suspending functions such as [collect], [single], [reduce], [toList], etc. + * or [launchIn] operator that starts collection of the flow in the given scope. + * They are applied to the upstream flow and trigger execution of all operations. + * Execution of the flow is also called _collecting the flow_ and is always performed in a suspending manner + * without actual blocking. Terminal operators complete normally or exceptionally depending on successful or failed + * execution of all the flow operations in the upstream. The most basic terminal operator is [collect], for example: + * + * ``` + * try { + * flow.collect { value -> + * println("Received $value") + * } + * } catch (e: Exception) { + * println("The flow has thrown an exception: $e") + * } + * ``` + * + * By default, flows are _sequential_ and all flow operations are executed sequentially in the same coroutine, + * with an exception for a few operations specifically designed to introduce concurrency into flow + * execution such as [buffer] and [flatMapMerge]. See their documentation for details. + * + * The `Flow` interface does not carry information whether a flow is a _cold_ stream that can be collected repeatedly and + * triggers execution of the same code every time it is collected, or if it is a _hot_ stream that emits different + * values from the same running source on each collection. Usually flows represent _cold_ streams, but + * there is a [SharedFlow] subtype that represents _hot_ streams. In addition to that, any flow can be turned + * into a _hot_ one by the [stateIn] and [shareIn] operators, or by converting the flow into a hot channel + * via the [produceIn] operator. + * + * ### Flow builders + * + * There are the following basic ways to create a flow: + * + * - [flowOf(...)][flowOf] functions to create a flow from a fixed set of values. + * - [asFlow()][asFlow] extension functions on various types to convert them into flows. + * - [flow { ... }][flow] builder function to construct arbitrary flows from + * sequential calls to [emit][FlowCollector.emit] function. + * - [channelFlow { ... }][channelFlow] builder function to construct arbitrary flows from + * potentially concurrent calls to the [send][kotlinx.coroutines.channels.SendChannel.send] function. + * - [MutableStateFlow] and [MutableSharedFlow] define the corresponding constructor functions to create + * a _hot_ flow that can be directly updated. + * + * ### Flow constraints + * + * All implementations of the `Flow` interface must adhere to two key properties described in detail below: + * + * - Context preservation. + * - Exception transparency. + * + * These properties ensure the ability to perform local reasoning about the code with flows and modularize the code + * in such a way that upstream flow emitters can be developed separately from downstream flow collectors. + * A user of a flow does not need to be aware of implementation details of the upstream flows it uses. + * + * ### Context preservation + * + * The flow has a context preservation property: it encapsulates its own execution context and never propagates or leaks + * it downstream, thus making reasoning about the execution context of particular transformations or terminal + * operations trivial. + * + * There is only one way to change the context of a flow: the [flowOn][Flow.flowOn] operator + * that changes the upstream context ("everything above the `flowOn` operator"). + * For additional information refer to its documentation. + * + * This reasoning can be demonstrated in practice: + * + * ``` + * val flowA = flowOf(1, 2, 3) + * .map { it + 1 } // Will be executed in ctxA + * .flowOn(ctxA) // Changes the upstream context: flowOf and map + * + * // Now we have a context-preserving flow: it is executed somewhere but this information is encapsulated in the flow itself + * + * val filtered = flowA // ctxA is encapsulated in flowA + * .filter { it == 3 } // Pure operator without a context yet + * + * withContext(Dispatchers.Main) { + * // All non-encapsulated operators will be executed in Main: filter and single + * val result = filtered.single() + * myUi.text = result + * } + * ``` + * + * From the implementation point of view, it means that all flow implementations should + * only emit from the same coroutine context. + * This constraint is efficiently enforced by the default [flow] builder. + * The [flow] builder should be used if the flow implementation does not start any coroutines. + * Its implementation prevents most of the development mistakes: + * + * ``` + * val myFlow = flow { + * // GlobalScope.launch { // is prohibited + * // launch(Dispatchers.IO) { // is prohibited + * // withContext(CoroutineName("myFlow")) { // is prohibited + * emit(1) // OK + * coroutineScope { + * emit(2) // OK -- still the same coroutine + * } + * } + * ``` + * + * Use [channelFlow] if the collection and emission of a flow are to be separated into multiple coroutines. + * It encapsulates all the context preservation work and allows you to focus on your + * domain-specific problem, rather than invariant implementation details. + * It is possible to use any combination of coroutine builders from within [channelFlow]. + * + * If you are looking for performance and are sure that no concurrent emits and context jumps will happen, + * the [flow] builder can be used alongside a [coroutineScope] or [supervisorScope] instead: + * - Scoped primitive should be used to provide a [CoroutineScope]. + * - Changing the context of emission is prohibited, no matter whether it is `withContext(ctx)` or + * a builder argument (e.g. `launch(ctx)`). + * - Collecting another flow from a separate context is allowed, but it has the same effect as + * applying the [flowOn] operator to that flow, which is more efficient. + * + * ### Exception transparency + * + * When `emit` or `emitAll` throws, the Flow implementations must immediately stop emitting new values and finish with an exception. + * For diagnostics or application-specific purposes, the exception may be different from the one thrown by the emit operation, + * suppressing the original exception as discussed below. + * If there is a need to emit values after the downstream failed, please use the [catch][Flow.catch] operator. + * + * The [catch][Flow.catch] operator only catches upstream exceptions, but passes + * all downstream exceptions. Similarly, terminal operators like [collect][Flow.collect] + * throw any unhandled exceptions that occur in their code or in upstream flows, for example: + * + * ``` + * flow { emitData() } + * .map { computeOne(it) } + * .catch { ... } // catches exceptions in emitData and computeOne + * .map { computeTwo(it) } + * .collect { process(it) } // throws exceptions from process and computeTwo + * ``` + * The same reasoning can be applied to the [onCompletion] operator that is a declarative replacement for the `finally` block. + * + * All exception-handling Flow operators follow the principle of exception suppression: + * + * If the upstream flow throws an exception during its completion when the downstream exception has been thrown, + * the downstream exception becomes superseded and suppressed by the upstream exception, being a semantic + * equivalent of throwing from `finally` block. However, this doesn't affect the operation of the exception-handling operators, + * which consider the downstream exception to be the root cause and behave as if the upstream didn't throw anything. + * + * Failure to adhere to the exception transparency requirement can lead to strange behaviors which make + * it hard to reason about the code because an exception in the `collect { ... }` could be somehow "caught" + * by an upstream flow, limiting the ability of local reasoning about the code. + * + * Flow machinery enforces exception transparency at runtime and throws [IllegalStateException] on any attempt to emit a value, + * if an exception has been thrown on previous attempt. + * + * ### Reactive streams + * + * 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. + * + * ### Not stable for inheritance + * + * **The `Flow` interface is not stable for inheritance in 3rd party libraries**, as new methods + * might be added to this interface in the future, but is stable for use. + * + * Use the `flow { ... }` builder function to create an implementation, or extend [AbstractFlow]. + * These implementations ensure that the context preservation property is not violated, and prevent most + * of the developer mistakes related to concurrency, inconsistent flow dispatchers, and cancellation. + */ +public actual interface Flow { + + /** + * Accepts the given [collector] and [emits][FlowCollector.emit] values into it. + * + * This method can be used along with SAM-conversion of [FlowCollector]: + * ``` + * myFlow.collect { value -> println("Collected $value") } + * ``` + * + * ### Method inheritance + * + * To ensure the context preservation property, it is not recommended implementing this method directly. + * Instead, [AbstractFlow] can be used as the base type to properly ensure flow's properties. + * + * All default flow implementations ensure context preservation and exception transparency properties on a best-effort basis + * and throw [IllegalStateException] if a violation was detected. + */ + public actual suspend fun collect(collector: FlowCollector) +} diff --git a/kotlinx-coroutines-core/wasmWasi/src/flow/Flow.wasm.kt b/kotlinx-coroutines-core/wasmWasi/src/flow/Flow.wasm.kt new file mode 100644 index 0000000000..a8c3ccb92f --- /dev/null +++ b/kotlinx-coroutines-core/wasmWasi/src/flow/Flow.wasm.kt @@ -0,0 +1,195 @@ +package kotlinx.coroutines.flow + +import kotlinx.coroutines.* +import kotlinx.coroutines.flow.internal.* +import kotlin.coroutines.* + +/** + * An asynchronous data stream that sequentially emits values and completes normally or with an exception. + * + * _Intermediate operators_ on the flow such as [map], [filter], [take], [zip], etc are functions that are + * applied to the _upstream_ flow or flows and return a _downstream_ flow where further operators can be applied to. + * Intermediate operations do not execute any code in the flow and are not suspending functions themselves. + * They only set up a chain of operations for future execution and quickly return. + * This is known as a _cold flow_ property. + * + * _Terminal operators_ on the flow are either suspending functions such as [collect], [single], [reduce], [toList], etc. + * or [launchIn] operator that starts collection of the flow in the given scope. + * They are applied to the upstream flow and trigger execution of all operations. + * Execution of the flow is also called _collecting the flow_ and is always performed in a suspending manner + * without actual blocking. Terminal operators complete normally or exceptionally depending on successful or failed + * execution of all the flow operations in the upstream. The most basic terminal operator is [collect], for example: + * + * ``` + * try { + * flow.collect { value -> + * println("Received $value") + * } + * } catch (e: Exception) { + * println("The flow has thrown an exception: $e") + * } + * ``` + * + * By default, flows are _sequential_ and all flow operations are executed sequentially in the same coroutine, + * with an exception for a few operations specifically designed to introduce concurrency into flow + * execution such as [buffer] and [flatMapMerge]. See their documentation for details. + * + * The `Flow` interface does not carry information whether a flow is a _cold_ stream that can be collected repeatedly and + * triggers execution of the same code every time it is collected, or if it is a _hot_ stream that emits different + * values from the same running source on each collection. Usually flows represent _cold_ streams, but + * there is a [SharedFlow] subtype that represents _hot_ streams. In addition to that, any flow can be turned + * into a _hot_ one by the [stateIn] and [shareIn] operators, or by converting the flow into a hot channel + * via the [produceIn] operator. + * + * ### Flow builders + * + * There are the following basic ways to create a flow: + * + * - [flowOf(...)][flowOf] functions to create a flow from a fixed set of values. + * - [asFlow()][asFlow] extension functions on various types to convert them into flows. + * - [flow { ... }][flow] builder function to construct arbitrary flows from + * sequential calls to [emit][FlowCollector.emit] function. + * - [channelFlow { ... }][channelFlow] builder function to construct arbitrary flows from + * potentially concurrent calls to the [send][kotlinx.coroutines.channels.SendChannel.send] function. + * - [MutableStateFlow] and [MutableSharedFlow] define the corresponding constructor functions to create + * a _hot_ flow that can be directly updated. + * + * ### Flow constraints + * + * All implementations of the `Flow` interface must adhere to two key properties described in detail below: + * + * - Context preservation. + * - Exception transparency. + * + * These properties ensure the ability to perform local reasoning about the code with flows and modularize the code + * in such a way that upstream flow emitters can be developed separately from downstream flow collectors. + * A user of a flow does not need to be aware of implementation details of the upstream flows it uses. + * + * ### Context preservation + * + * The flow has a context preservation property: it encapsulates its own execution context and never propagates or leaks + * it downstream, thus making reasoning about the execution context of particular transformations or terminal + * operations trivial. + * + * There is only one way to change the context of a flow: the [flowOn][Flow.flowOn] operator + * that changes the upstream context ("everything above the `flowOn` operator"). + * For additional information refer to its documentation. + * + * This reasoning can be demonstrated in practice: + * + * ``` + * val flowA = flowOf(1, 2, 3) + * .map { it + 1 } // Will be executed in ctxA + * .flowOn(ctxA) // Changes the upstream context: flowOf and map + * + * // Now we have a context-preserving flow: it is executed somewhere but this information is encapsulated in the flow itself + * + * val filtered = flowA // ctxA is encapsulated in flowA + * .filter { it == 3 } // Pure operator without a context yet + * + * withContext(Dispatchers.Main) { + * // All non-encapsulated operators will be executed in Main: filter and single + * val result = filtered.single() + * myUi.text = result + * } + * ``` + * + * From the implementation point of view, it means that all flow implementations should + * only emit from the same coroutine context. + * This constraint is efficiently enforced by the default [flow] builder. + * The [flow] builder should be used if the flow implementation does not start any coroutines. + * Its implementation prevents most of the development mistakes: + * + * ``` + * val myFlow = flow { + * // GlobalScope.launch { // is prohibited + * // launch(Dispatchers.IO) { // is prohibited + * // withContext(CoroutineName("myFlow")) { // is prohibited + * emit(1) // OK + * coroutineScope { + * emit(2) // OK -- still the same coroutine + * } + * } + * ``` + * + * Use [channelFlow] if the collection and emission of a flow are to be separated into multiple coroutines. + * It encapsulates all the context preservation work and allows you to focus on your + * domain-specific problem, rather than invariant implementation details. + * It is possible to use any combination of coroutine builders from within [channelFlow]. + * + * If you are looking for performance and are sure that no concurrent emits and context jumps will happen, + * the [flow] builder can be used alongside a [coroutineScope] or [supervisorScope] instead: + * - Scoped primitive should be used to provide a [CoroutineScope]. + * - Changing the context of emission is prohibited, no matter whether it is `withContext(ctx)` or + * a builder argument (e.g. `launch(ctx)`). + * - Collecting another flow from a separate context is allowed, but it has the same effect as + * applying the [flowOn] operator to that flow, which is more efficient. + * + * ### Exception transparency + * + * When `emit` or `emitAll` throws, the Flow implementations must immediately stop emitting new values and finish with an exception. + * For diagnostics or application-specific purposes, the exception may be different from the one thrown by the emit operation, + * suppressing the original exception as discussed below. + * If there is a need to emit values after the downstream failed, please use the [catch][Flow.catch] operator. + * + * The [catch][Flow.catch] operator only catches upstream exceptions, but passes + * all downstream exceptions. Similarly, terminal operators like [collect][Flow.collect] + * throw any unhandled exceptions that occur in their code or in upstream flows, for example: + * + * ``` + * flow { emitData() } + * .map { computeOne(it) } + * .catch { ... } // catches exceptions in emitData and computeOne + * .map { computeTwo(it) } + * .collect { process(it) } // throws exceptions from process and computeTwo + * ``` + * The same reasoning can be applied to the [onCompletion] operator that is a declarative replacement for the `finally` block. + * + * All exception-handling Flow operators follow the principle of exception suppression: + * + * If the upstream flow throws an exception during its completion when the downstream exception has been thrown, + * the downstream exception becomes superseded and suppressed by the upstream exception, being a semantic + * equivalent of throwing from `finally` block. However, this doesn't affect the operation of the exception-handling operators, + * which consider the downstream exception to be the root cause and behave as if the upstream didn't throw anything. + * + * Failure to adhere to the exception transparency requirement can lead to strange behaviors which make + * it hard to reason about the code because an exception in the `collect { ... }` could be somehow "caught" + * by an upstream flow, limiting the ability of local reasoning about the code. + * + * Flow machinery enforces exception transparency at runtime and throws [IllegalStateException] on any attempt to emit a value, + * if an exception has been thrown on previous attempt. + * + * ### Reactive streams + * + * 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. + * + * ### Not stable for inheritance + * + * **The `Flow` interface is not stable for inheritance in 3rd party libraries**, as new methods + * might be added to this interface in the future, but is stable for use. + * + * Use the `flow { ... }` builder function to create an implementation, or extend [AbstractFlow]. + * These implementations ensure that the context preservation property is not violated, and prevent most + * of the developer mistakes related to concurrency, inconsistent flow dispatchers, and cancellation. + */ +public actual interface Flow { + + /** + * Accepts the given [collector] and [emits][FlowCollector.emit] values into it. + * + * This method can be used along with SAM-conversion of [FlowCollector]: + * ``` + * myFlow.collect { value -> println("Collected $value") } + * ``` + * + * ### Method inheritance + * + * To ensure the context preservation property, it is not recommended implementing this method directly. + * Instead, [AbstractFlow] can be used as the base type to properly ensure flow's properties. + * + * All default flow implementations ensure context preservation and exception transparency properties on a best-effort basis + * and throw [IllegalStateException] if a violation was detected. + */ + public actual suspend fun collect(collector: FlowCollector) +} From 60e4c307ae5d1eb888591e30cd6bd22dd67ecab2 Mon Sep 17 00:00:00 2001 From: Artem Kobzar Date: Tue, 7 Jul 2026 19:00:27 +0200 Subject: [PATCH 2/5] fixup! [K/JS] Add exportability of Flow with a conversion method to `AsyncIterator` --- .../api/kotlinx-coroutines-core.klib.api | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.klib.api b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.klib.api index 5317dbcb48..fb42c20fe4 100644 --- a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.klib.api +++ b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.klib.api @@ -198,6 +198,19 @@ abstract interface <#A: out kotlin/Any?> kotlinx.coroutines.channels/ChannelIter abstract interface <#A: out kotlin/Any?> kotlinx.coroutines.flow/Flow { // kotlinx.coroutines.flow/Flow|null[0] abstract suspend fun collect(kotlinx.coroutines.flow/FlowCollector<#A>) // kotlinx.coroutines.flow/Flow.collect|collect(kotlinx.coroutines.flow.FlowCollector<1:0>){}[0] + + // Targets: [js] + open fun asAsyncIterable(): kotlinx.coroutines.internal/JsAsyncIterable<#A> // kotlinx.coroutines.flow/Flow.asAsyncIterable|asAsyncIterable(){}[0] + + // Targets: [js] + open fun asAsyncIterable(kotlinx.coroutines/CoroutineScope): kotlinx.coroutines.internal/JsAsyncIterable<#A> // kotlinx.coroutines.flow/Flow.asAsyncIterable|asAsyncIterable(kotlinx.coroutines.CoroutineScope){}[0] + + // Targets: [js] + final object Companion { // kotlinx.coroutines.flow/Flow.Companion|null[0] + final fun <#A2: kotlin/Any?> from(kotlin/Function0>): kotlinx.coroutines.flow/Flow<#A2> // kotlinx.coroutines.flow/Flow.Companion.from|from(kotlin.Function0>){0§}[0] + final fun <#A2: kotlin/Any?> from(kotlinx.coroutines.internal/JsAsyncIterable<#A2>): kotlinx.coroutines.flow/Flow<#A2> // kotlinx.coroutines.flow/Flow.Companion.from|from(kotlinx.coroutines.internal.JsAsyncIterable<0:0>){0§}[0] + final fun <#A2: kotlin/Any?> from(kotlinx.coroutines.internal/JsAsyncIterator<#A2>): kotlinx.coroutines.flow/Flow<#A2> // kotlinx.coroutines.flow/Flow.Companion.from|from(kotlinx.coroutines.internal.JsAsyncIterator<0:0>){0§}[0] + } } abstract interface <#A: out kotlin/Any?> kotlinx.coroutines.flow/SharedFlow : kotlinx.coroutines.flow/Flow<#A> { // kotlinx.coroutines.flow/SharedFlow|null[0] @@ -1263,7 +1276,7 @@ final inline fun <#A: kotlin/Any?> kotlinx.coroutines.internal/synchronized(kotl final inline fun <#A: kotlin/Any?> kotlinx.coroutines.internal/synchronizedImpl(kotlinx.coroutines.internal/SynchronizedObject, kotlin/Function0<#A>): #A // kotlinx.coroutines.internal/synchronizedImpl|synchronizedImpl(kotlinx.coroutines.internal.SynchronizedObject;kotlin.Function0<0:0>){0§}[0] // Targets: [js] -abstract interface <#A: out kotlin/Any?> kotlinx.coroutines.channels/ReceiveChannel : kotlinx.coroutines.channels/JsAsyncIterable<#A> { // kotlinx.coroutines.channels/ReceiveChannel|null[0] +abstract interface <#A: out kotlin/Any?> kotlinx.coroutines.channels/ReceiveChannel : kotlinx.coroutines.internal/JsAsyncIterable<#A> { // kotlinx.coroutines.channels/ReceiveChannel|null[0] abstract val isClosedForReceive // kotlinx.coroutines.channels/ReceiveChannel.isClosedForReceive|{}isClosedForReceive[0] abstract fun (): kotlin/Boolean // kotlinx.coroutines.channels/ReceiveChannel.isClosedForReceive.|(){}[0] abstract val isEmpty // kotlinx.coroutines.channels/ReceiveChannel.isEmpty|{}isEmpty[0] @@ -1309,10 +1322,10 @@ final inline fun <#A: out kotlin/Any?> kotlinx.coroutines.channels/kotlinx_corou final inline fun <#A: out kotlin/Any?> kotlinx.coroutines.channels/kotlinx_coroutines_channels_JsAsyncIterator_Companion_btfdib_invoke_jkqnwo(noinline kotlin/Function0>>, noinline kotlin/Function1<#A?, kotlin.js/Promise>>, noinline kotlin/Function1>>): kotlinx.coroutines.channels/JsAsyncIterator<#A> // kotlinx.coroutines.channels/kotlinx_coroutines_channels_JsAsyncIterator_Companion_btfdib_invoke_jkqnwo|kotlinx_coroutines_channels_JsAsyncIterator_Companion_btfdib_invoke_jkqnwo(kotlin.Function0>>;kotlin.Function1<0:0?,kotlin.js.Promise>>;kotlin.Function1>>){0§}[0] // Targets: [js] -final inline fun <#A: out kotlin/Any?> kotlinx.coroutines.channels/kotlinx_coroutines_channels_JsIteratorResult_Companion_6yc9qk_copy_1tks5(noinline kotlinx.coroutines.channels/JsIteratorResult<#A>, noinline #A? = ..., noinline kotlin/Boolean = ...): kotlinx.coroutines.channels/JsIteratorResult<#A> // kotlinx.coroutines.channels/kotlinx_coroutines_channels_JsIteratorResult_Companion_6yc9qk_copy_1tks5|kotlinx_coroutines_channels_JsIteratorResult_Companion_6yc9qk_copy_1tks5(kotlinx.coroutines.channels.JsIteratorResult<0:0>;0:0?;kotlin.Boolean){0§}[0] +final inline fun <#A: out kotlin/Any?> kotlinx.coroutines.internal/kotlinx_coroutines_internal_JsIteratorResult_Companion_kfjag7_copy_1tks5(noinline kotlinx.coroutines.internal/JsIteratorResult<#A>, noinline #A? = ..., noinline kotlin/Boolean = ...): kotlinx.coroutines.internal/JsIteratorResult<#A> // kotlinx.coroutines.internal/kotlinx_coroutines_internal_JsIteratorResult_Companion_kfjag7_copy_1tks5|kotlinx_coroutines_internal_JsIteratorResult_Companion_kfjag7_copy_1tks5(kotlinx.coroutines.internal.JsIteratorResult<0:0>;0:0?;kotlin.Boolean){0§}[0] // Targets: [js] -final inline fun <#A: out kotlin/Any?> kotlinx.coroutines.channels/kotlinx_coroutines_channels_JsIteratorResult_Companion_6yc9qk_invoke_jkqnwo(noinline #A? = ..., noinline kotlin/Boolean): kotlinx.coroutines.channels/JsIteratorResult<#A> // kotlinx.coroutines.channels/kotlinx_coroutines_channels_JsIteratorResult_Companion_6yc9qk_invoke_jkqnwo|kotlinx_coroutines_channels_JsIteratorResult_Companion_6yc9qk_invoke_jkqnwo(0:0?;kotlin.Boolean){0§}[0] +final inline fun <#A: out kotlin/Any?> kotlinx.coroutines.internal/kotlinx_coroutines_internal_JsIteratorResult_Companion_kfjag7_invoke_jkqnwo(noinline #A? = ..., noinline kotlin/Boolean): kotlinx.coroutines.internal/JsIteratorResult<#A> // kotlinx.coroutines.internal/kotlinx_coroutines_internal_JsIteratorResult_Companion_kfjag7_invoke_jkqnwo|kotlinx_coroutines_internal_JsIteratorResult_Companion_kfjag7_invoke_jkqnwo(0:0?;kotlin.Boolean){0§}[0] // Targets: [js] final suspend fun (org.w3c.dom/Window).kotlinx.coroutines/awaitAnimationFrame(): kotlin/Double // kotlinx.coroutines/awaitAnimationFrame|awaitAnimationFrame@org.w3c.dom.Window(){}[0] From 1229ff8719be35ce162c4bf470d62a04e260b179 Mon Sep 17 00:00:00 2001 From: Artem Kobzar Date: Tue, 4 Aug 2026 16:10:35 +0200 Subject: [PATCH 3/5] fixup! [K/JS] Add exportability of Flow with a conversion method to `AsyncIterator` --- .../js/src/flow/Flow.js.kt | 14 +- .../js/test/ChannelInteropTest.kt | 12 +- .../js/test/FlowInteropTest.kt | 210 +++--------------- .../js/test/internal/jsIteratorAssertions.kt | 14 ++ 4 files changed, 53 insertions(+), 197 deletions(-) create mode 100644 kotlinx-coroutines-core/js/test/internal/jsIteratorAssertions.kt diff --git a/kotlinx-coroutines-core/js/src/flow/Flow.js.kt b/kotlinx-coroutines-core/js/src/flow/Flow.js.kt index 741a51d8ed..bf015a3716 100644 --- a/kotlinx-coroutines-core/js/src/flow/Flow.js.kt +++ b/kotlinx-coroutines-core/js/src/flow/Flow.js.kt @@ -5,7 +5,10 @@ package kotlinx.coroutines.flow import kotlinx.coroutines.* import kotlinx.coroutines.internal.JsAsyncIterable import kotlinx.coroutines.internal.JsAsyncIterator +import kotlinx.coroutines.internal.JsIteratorResult +import kotlinx.coroutines.internal.JsOptionalExport import kotlin.coroutines.EmptyCoroutineContext +import kotlin.js.Promise /** * An asynchronous data stream that sequentially emits values and completes normally or with an exception. @@ -176,6 +179,7 @@ import kotlin.coroutines.EmptyCoroutineContext * These implementations ensure that the context preservation property is not violated, and prevent most * of the developer mistakes related to concurrency, inconsistent flow dispatchers, and cancellation. */ +@JsOptionalExport(couldBeConvertedToExplicitExport = true) public actual interface Flow { /** @@ -198,14 +202,8 @@ public actual interface Flow { public actual suspend fun collect(collector: FlowCollector) - - @JsExport.Ignore - // For Kotlin side only to be able to set up a custom scope for the iterator - public fun asAsyncIterable(scope: CoroutineScope): JsAsyncIterable = - buffer(0).produceIn(scope) - public fun asAsyncIterable(): JsAsyncIterable = - asAsyncIterable(CoroutineScope(EmptyCoroutineContext)) + buffer(0).produceIn(GlobalScope) @JsExport.Ignore // Important note: it would be much nicer to place those factory functions outside of Flow @@ -248,7 +246,7 @@ public actual interface Flow { } } finally { if (!completed) { - iterator.`return`().await() + iterator.asDynamic().`return`().unsafeCast>().await() } } } diff --git a/kotlinx-coroutines-core/js/test/ChannelInteropTest.kt b/kotlinx-coroutines-core/js/test/ChannelInteropTest.kt index 380ff0b964..69f9646c61 100644 --- a/kotlinx-coroutines-core/js/test/ChannelInteropTest.kt +++ b/kotlinx-coroutines-core/js/test/ChannelInteropTest.kt @@ -2,6 +2,8 @@ package kotlinx.coroutines import kotlinx.coroutines.channels.* import kotlinx.coroutines.internal.JsAsyncIterator +import kotlinx.coroutines.internal.JsIteratorResult +import kotlinx.coroutines.internal.assertNextStepToBe import kotlinx.coroutines.testing.* import kotlin.js.* import kotlin.test.* @@ -196,14 +198,4 @@ class ChannelInteropTest : TestBase() { assertNextStepToBe(iterator, done = true) assertEquals(2, channel.receive()) } - - private suspend fun assertNextStepToBe( - iterator: JsAsyncIterator, - value: T? = js("undefined"), - done: Boolean = false - ) { - val result = iterator.next().await() - assertEquals(done, result.done) - assertEquals(value, result.value) - } } diff --git a/kotlinx-coroutines-core/js/test/FlowInteropTest.kt b/kotlinx-coroutines-core/js/test/FlowInteropTest.kt index dcf6f5274f..0160b2dfac 100644 --- a/kotlinx-coroutines-core/js/test/FlowInteropTest.kt +++ b/kotlinx-coroutines-core/js/test/FlowInteropTest.kt @@ -7,6 +7,7 @@ import kotlin.js.* import kotlin.test.* import kotlinx.coroutines.internal.JsAsyncIterator import kotlinx.coroutines.internal.JsIteratorResult +import kotlinx.coroutines.internal.assertNextStepToBe class FlowInteropTest : TestBase() { @@ -16,43 +17,25 @@ class FlowInteropTest : TestBase() { fun testFlowToAsyncIteratorBasic() = runTest { val flow = flowOf(1, 2, 3) val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() - - val result1 = iterator.next().await() - assertEquals(1, result1.value) - assertEquals(false, result1.done) - - val result2 = iterator.next().await() - assertEquals(2, result2.value) - assertEquals(false, result2.done) - - val result3 = iterator.next().await() - assertEquals(3, result3.value) - assertEquals(false, result3.done) - - val result4 = iterator.next().await() - assertEquals(true, result4.done) + assertNextStepToBe(iterator, 1, done = false) + assertNextStepToBe(iterator, 2, done = false) + assertNextStepToBe(iterator, 3, done = false) + assertNextStepToBe(iterator, done = true) } @Test fun testFlowToAsyncIteratorEmpty() = runTest { val flow = emptyFlow() val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() - - val result = iterator.next().await() - assertEquals(true, result.done) + assertNextStepToBe(iterator, done = true) } @Test fun testFlowToAsyncIteratorSingle() = runTest { val flow = flowOf(42) val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() - - val result1 = iterator.next().await() - assertEquals(42, result1.value) - assertEquals(false, result1.done) - - val result2 = iterator.next().await() - assertEquals(true, result2.done) + assertNextStepToBe(iterator, 42, done = false) + assertNextStepToBe(iterator, done = true) } @Test @@ -65,17 +48,11 @@ class FlowInteropTest : TestBase() { emit(5) } val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() - - val result1 = iterator.next().await() - assertEquals(1, result1.value) - + assertNextStepToBe(iterator, 1, done = false) // Call return() to stop iteration early - val returnResult = iterator.`return`().await() + val returnResult = iterator.asDynamic().`return`().unsafeCast>>().await() assertEquals(true, returnResult.done) - - // Next calls should return done - val result2 = iterator.next().await() - assertEquals(true, result2.done) + assertNextStepToBe(iterator, done = true) } @Test @@ -86,18 +63,12 @@ class FlowInteropTest : TestBase() { emit(3) } val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() - - val result1 = iterator.next().await() - assertEquals(1, result1.value) - + assertNextStepToBe(iterator, 1, done = false) // Call throw() to cancel the iterator val error = js("new Error('test error')") - try { - iterator.`throw`(error).await() - fail("Should have thrown") - } catch (e: Throwable) { - // Expected - } + assertFailsWith { iterator.`throw`(error).await() } + .apply { assertEquals("test error", message) } + assertNextStepToBe(iterator, done = true) } @Test @@ -108,20 +79,11 @@ class FlowInteropTest : TestBase() { throw TestException("test exception") } val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() - - val result1 = iterator.next().await() - assertEquals(1, result1.value) - - val result2 = iterator.next().await() - assertEquals(2, result2.value) - + assertNextStepToBe(iterator, 1, done = false) + assertNextStepToBe(iterator, 2, done = false) // Next call should throw the exception - try { - val result = iterator.next().await() - fail("Should have thrown TestException, but return result ${result.value}") - } catch (e: TestException) { - assertEquals("test exception", e.message) - } + assertFailsWith { iterator.next().await() } + .apply { assertEquals("test exception", message) } } @Test @@ -129,17 +91,10 @@ class FlowInteropTest : TestBase() { val flow = flowOf(1, 2, 3) .map { it * 2 } .filter { it > 2 } - val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() - - val result1 = iterator.next().await() - assertEquals(4, result1.value) - - val result2 = iterator.next().await() - assertEquals(6, result2.value) - - val result3 = iterator.next().await() - assertEquals(true, result3.done) + assertNextStepToBe(iterator, 4, done = false) + assertNextStepToBe(iterator, 6, done = false) + assertNextStepToBe(iterator, done = true) } @Test @@ -152,62 +107,12 @@ class FlowInteropTest : TestBase() { emit(5) } val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() - - val result1 = iterator.next().await() - assertEquals(1, result1.value) - assertEquals(false, result1.done) - - val result2 = iterator.next().await() - assertEquals(2, result2.value) - assertEquals(false, result2.done) - - val returnResult = iterator.`return`().await() + assertNextStepToBe(iterator, 1, done = false) + assertNextStepToBe(iterator, 2, done = false) + val returnResult = iterator.`return`(42).await() assertEquals(true, returnResult.done) - - val result3 = iterator.next().await() - assertEquals(true, result3.done) - - val result4 = iterator.next().await() - assertEquals(true, result4.done) - } - - @Test - fun testFlowToAsyncIteratorCancellationExceptionReturnsDone() = runTest { - val flow = flow { - emit(1) - emit(2) - emit(3) - throw CancellationException("flow cancelled") - emit(4) // These should never be emitted - emit(5) - emit(6) - } - val iterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() - - val result1 = iterator.next().await() - assertEquals(1, result1.value) - assertEquals(false, result1.done) - - val result2 = iterator.next().await() - assertEquals(2, result2.value) - assertEquals(false, result2.done) - - val result3 = iterator.next().await() - assertEquals(3, result3.value) - assertEquals(false, result3.done) - - // When flow throws CancellationException, next() should return done: true - // without propagating the exception, and unconditionally stop emitting - val result4 = iterator.next().await() - assertEquals(true, result4.done) - - // Subsequent calls should also return done: true, ensuring elements after - // CancellationException are never emitted - val result5 = iterator.next().await() - assertEquals(true, result5.done) - - val result6 = iterator.next().await() - assertEquals(true, result6.done) + assertNextStepToBe(iterator, done = true) + assertNextStepToBe(iterator, done = true) } // ===== AsyncIterator to Flow tests ===== @@ -216,10 +121,8 @@ class FlowInteropTest : TestBase() { fun testAsyncIteratorToFlowBasic() = runTest { val asyncIterator = createAsyncIterator(listOf(1, 2, 3)) val flow = Flow.from(asyncIterator) - val results = mutableListOf() flow.collect { results.add(it) } - assertEquals(listOf(1, 2, 3), results) } @@ -227,10 +130,8 @@ class FlowInteropTest : TestBase() { fun testAsyncIteratorToFlowEmpty() = runTest { val asyncIterator = createAsyncIterator(emptyList()) val flow = Flow.from(asyncIterator) - val results = mutableListOf() flow.collect { results.add(it) } - assertEquals(emptyList(), results) } @@ -238,10 +139,8 @@ class FlowInteropTest : TestBase() { fun testAsyncIteratorToFlowSingle() = runTest { val asyncIterator = createAsyncIterator(listOf(42)) val flow = Flow.from(asyncIterator) - val results = mutableListOf() flow.collect { results.add(it) } - assertEquals(listOf(42), results) } @@ -253,10 +152,8 @@ class FlowInteropTest : TestBase() { onReturn = { returnCalled = true } ) val flow = Flow.from(asyncIterator) - val results = mutableListOf() flow.take(2).collect { results.add(it) } - assertEquals(listOf(1, 2), results) yield() // Allow cleanup to happen assertTrue(returnCalled, "return() should be called on cancellation") @@ -269,7 +166,6 @@ class FlowInteropTest : TestBase() { TestException("iterator error") ) val flow = Flow.from(asyncIterator) - val results = mutableListOf() try { flow.collect { results.add(it) } @@ -277,7 +173,6 @@ class FlowInteropTest : TestBase() { } catch (e: TestException) { assertEquals("iterator error", e.message) } - assertEquals(listOf(1, 2), results) } @@ -286,9 +181,7 @@ class FlowInteropTest : TestBase() { val generator: () -> JsAsyncIterator = { createAsyncIterator(listOf(10, 20, 30)) } - val flow = Flow.from(generator) - val results = mutableListOf() flow.collect { results.add(it) } @@ -299,10 +192,8 @@ class FlowInteropTest : TestBase() { fun testAsyncIterableToFlow() = runTest { val asyncIterable = createAsyncIterable(listOf(5, 10, 15)) val flow = Flow.from(asyncIterable) - val results = mutableListOf() flow.collect { results.add(it) } - assertEquals(listOf(5, 10, 15), results) } @@ -312,10 +203,8 @@ class FlowInteropTest : TestBase() { val flow = Flow.from(asyncIterator) .map { it * 2 } .filter { it > 5 } - val results = mutableListOf() flow.collect { results.add(it) } - assertEquals(listOf(6, 8, 10), results) } @@ -324,10 +213,8 @@ class FlowInteropTest : TestBase() { fun testRoundTripFlowToAsyncIterableToFlow() = runTest { val originalFlow = flowOf(1, 2, 3, 4, 5) val convertedFlow = Flow.from(originalFlow.asAsyncIterable()) - val results = mutableListOf() convertedFlow.collect { results.add(it) } - assertEquals(listOf(1, 2, 3, 4, 5), results) } @@ -336,10 +223,8 @@ class FlowInteropTest : TestBase() { val originalFlow = flowOf(1, 2, 3, 4, 5) val iterator: JsAsyncIterator = originalFlow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() val convertedFlow = Flow.from(iterator) - val results = mutableListOf() convertedFlow.collect { results.add(it) } - assertEquals(listOf(1, 2, 3, 4, 5), results) } @@ -348,18 +233,10 @@ class FlowInteropTest : TestBase() { val originalIterator = createAsyncIterator(listOf(10, 20, 30)) val flow = Flow.from(originalIterator) val convertedIterator: JsAsyncIterator = flow.asAsyncIterable().asDynamic()[js("Symbol.asyncIterator")]() - - val result1 = convertedIterator.next().await() - assertEquals(10, result1.value) - - val result2 = convertedIterator.next().await() - assertEquals(20, result2.value) - - val result3 = convertedIterator.next().await() - assertEquals(30, result3.value) - - val result4 = convertedIterator.next().await() - assertEquals(true, result4.done) + assertNextStepToBe(convertedIterator, 10, done = false) + assertNextStepToBe(convertedIterator, 20, done = false) + assertNextStepToBe(convertedIterator, 30, done = false) + assertNextStepToBe(convertedIterator, done = true) } // ===== Helper functions ===== @@ -375,7 +252,6 @@ class FlowInteropTest : TestBase() { Promise.resolve(js("({ value: undefined, done: true })")) } } - iterator.`return` = fun(): Promise> { return Promise.resolve(js("({ value: undefined, done: true })")) } @@ -396,7 +272,6 @@ class FlowInteropTest : TestBase() { Promise.resolve(js("({ value: undefined, done: true })")) } } - iterator.`return` = fun(): Promise> { onReturn() return Promise.resolve(js("({ value: undefined, done: true })")) @@ -418,29 +293,6 @@ class FlowInteropTest : TestBase() { Promise.reject(exception) } } - - iterator.`return` = fun(): Promise> { - return Promise.resolve(js("({ value: undefined, done: true })")) - } - iterator.`throw` = fun(error: Throwable): Promise> { - return Promise.reject(error) - } - return iterator - } - - private fun createAsyncIteratorWithCallCounter(values: List, onNextCall: () -> Unit): JsAsyncIterator { - var index = 0 - val iterator = js("({})") - iterator.next = fun(): Promise> { - onNextCall() - return if (index < values.size) { - val value = values[index++] - Promise.resolve(js("({ value: value, done: false })")) - } else { - Promise.resolve(js("({ value: undefined, done: true })")) - } - } - iterator.`return` = fun(): Promise> { return Promise.resolve(js("({ value: undefined, done: true })")) } diff --git a/kotlinx-coroutines-core/js/test/internal/jsIteratorAssertions.kt b/kotlinx-coroutines-core/js/test/internal/jsIteratorAssertions.kt new file mode 100644 index 0000000000..e57ffea026 --- /dev/null +++ b/kotlinx-coroutines-core/js/test/internal/jsIteratorAssertions.kt @@ -0,0 +1,14 @@ +package kotlinx.coroutines.internal + +import kotlinx.coroutines.await +import kotlin.test.assertEquals + +internal suspend fun assertNextStepToBe( + iterator: JsAsyncIterator, + value: T? = js("undefined"), + done: Boolean = false +) { + val result = iterator.next().await() + assertEquals(done, result.done) + assertEquals(value, result.value) +} From 196f9062e55facbabc0af8aaf8e289b24a9c4210 Mon Sep 17 00:00:00 2001 From: Artem Kobzar Date: Fri, 7 Aug 2026 19:06:53 +0200 Subject: [PATCH 4/5] fixup! [K/JS] Add exportability of Flow with a conversion method to `AsyncIterator` --- .../api/kotlinx-coroutines-core.klib.api | 9 +- .../concurrent/src/flow/Flow.concurrent.kt | 190 ---------------- .../js/src/flow/Flow.js.kt | 210 +++--------------- .../js/test/FlowInteropTest.kt | 12 +- .../wasmJs/src/flow/Flow.wasm.kt | 190 ---------------- .../wasmWasi/src/flow/Flow.wasm.kt | 190 ---------------- 6 files changed, 40 insertions(+), 761 deletions(-) diff --git a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.klib.api b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.klib.api index fb42c20fe4..8b409e3f09 100644 --- a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.klib.api +++ b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.klib.api @@ -202,9 +202,6 @@ abstract interface <#A: out kotlin/Any?> kotlinx.coroutines.flow/Flow { // kotli // Targets: [js] open fun asAsyncIterable(): kotlinx.coroutines.internal/JsAsyncIterable<#A> // kotlinx.coroutines.flow/Flow.asAsyncIterable|asAsyncIterable(){}[0] - // Targets: [js] - open fun asAsyncIterable(kotlinx.coroutines/CoroutineScope): kotlinx.coroutines.internal/JsAsyncIterable<#A> // kotlinx.coroutines.flow/Flow.asAsyncIterable|asAsyncIterable(kotlinx.coroutines.CoroutineScope){}[0] - // Targets: [js] final object Companion { // kotlinx.coroutines.flow/Flow.Companion|null[0] final fun <#A2: kotlin/Any?> from(kotlin/Function0>): kotlinx.coroutines.flow/Flow<#A2> // kotlinx.coroutines.flow/Flow.Companion.from|from(kotlin.Function0>){0§}[0] @@ -1294,7 +1291,7 @@ abstract interface <#A: out kotlin/Any?> kotlinx.coroutines.channels/ReceiveChan abstract fun tryReceive(): kotlinx.coroutines.channels/ChannelResult<#A> // kotlinx.coroutines.channels/ReceiveChannel.tryReceive|tryReceive(){}[0] abstract suspend fun receive(): #A // kotlinx.coroutines.channels/ReceiveChannel.receive|receive(){}[0] abstract suspend fun receiveCatching(): kotlinx.coroutines.channels/ChannelResult<#A> // kotlinx.coroutines.channels/ReceiveChannel.receiveCatching|receiveCatching(){}[0] - open fun asyncIterator(): kotlinx.coroutines.channels/JsAsyncIterator<#A> // kotlinx.coroutines.channels/ReceiveChannel.asyncIterator|asyncIterator(){}[0] + open fun asyncIterator(): kotlinx.coroutines.internal/JsAsyncIterator<#A> // kotlinx.coroutines.channels/ReceiveChannel.asyncIterator|asyncIterator(){}[0] open fun cancel() // kotlinx.coroutines.channels/ReceiveChannel.cancel|cancel(){}[0] open fun poll(): #A? // kotlinx.coroutines.channels/ReceiveChannel.poll|poll(){}[0] open suspend fun receiveOrNull(): #A? // kotlinx.coroutines.channels/ReceiveChannel.receiveOrNull|receiveOrNull(){}[0] @@ -1316,10 +1313,10 @@ final fun <#A: kotlin/Any?> (kotlinx.coroutines/CoroutineScope).kotlinx.coroutin final fun <#A: kotlin/Any?> (kotlinx.coroutines/Deferred<#A>).kotlinx.coroutines/asPromise(): kotlin.js/Promise<#A> // kotlinx.coroutines/asPromise|asPromise@kotlinx.coroutines.Deferred<0:0>(){0§}[0] // Targets: [js] -final inline fun <#A: out kotlin/Any?> kotlinx.coroutines.channels/kotlinx_coroutines_channels_JsAsyncIterator_Companion_btfdib_copy_1tks5(noinline kotlinx.coroutines.channels/JsAsyncIterator<#A>, noinline kotlin/Function0>> = ..., noinline kotlin/Function1<#A?, kotlin.js/Promise>> = ..., noinline kotlin/Function1>> = ...): kotlinx.coroutines.channels/JsAsyncIterator<#A> // kotlinx.coroutines.channels/kotlinx_coroutines_channels_JsAsyncIterator_Companion_btfdib_copy_1tks5|kotlinx_coroutines_channels_JsAsyncIterator_Companion_btfdib_copy_1tks5(kotlinx.coroutines.channels.JsAsyncIterator<0:0>;kotlin.Function0>>;kotlin.Function1<0:0?,kotlin.js.Promise>>;kotlin.Function1>>){0§}[0] +final inline fun <#A: out kotlin/Any?> kotlinx.coroutines.internal/kotlinx_coroutines_internal_JsAsyncIterator_Companion_9orgqy_copy_1tks5(noinline kotlinx.coroutines.internal/JsAsyncIterator<#A>, noinline kotlin/Function0>> = ..., noinline kotlin/Function1<#A?, kotlin.js/Promise>> = ..., noinline kotlin/Function1>> = ...): kotlinx.coroutines.internal/JsAsyncIterator<#A> // kotlinx.coroutines.internal/kotlinx_coroutines_internal_JsAsyncIterator_Companion_9orgqy_copy_1tks5|kotlinx_coroutines_internal_JsAsyncIterator_Companion_9orgqy_copy_1tks5(kotlinx.coroutines.internal.JsAsyncIterator<0:0>;kotlin.Function0>>;kotlin.Function1<0:0?,kotlin.js.Promise>>;kotlin.Function1>>){0§}[0] // Targets: [js] -final inline fun <#A: out kotlin/Any?> kotlinx.coroutines.channels/kotlinx_coroutines_channels_JsAsyncIterator_Companion_btfdib_invoke_jkqnwo(noinline kotlin/Function0>>, noinline kotlin/Function1<#A?, kotlin.js/Promise>>, noinline kotlin/Function1>>): kotlinx.coroutines.channels/JsAsyncIterator<#A> // kotlinx.coroutines.channels/kotlinx_coroutines_channels_JsAsyncIterator_Companion_btfdib_invoke_jkqnwo|kotlinx_coroutines_channels_JsAsyncIterator_Companion_btfdib_invoke_jkqnwo(kotlin.Function0>>;kotlin.Function1<0:0?,kotlin.js.Promise>>;kotlin.Function1>>){0§}[0] +final inline fun <#A: out kotlin/Any?> kotlinx.coroutines.internal/kotlinx_coroutines_internal_JsAsyncIterator_Companion_9orgqy_invoke_jkqnwo(noinline kotlin/Function0>>, noinline kotlin/Function1<#A?, kotlin.js/Promise>>, noinline kotlin/Function1>>): kotlinx.coroutines.internal/JsAsyncIterator<#A> // kotlinx.coroutines.internal/kotlinx_coroutines_internal_JsAsyncIterator_Companion_9orgqy_invoke_jkqnwo|kotlinx_coroutines_internal_JsAsyncIterator_Companion_9orgqy_invoke_jkqnwo(kotlin.Function0>>;kotlin.Function1<0:0?,kotlin.js.Promise>>;kotlin.Function1>>){0§}[0] // Targets: [js] final inline fun <#A: out kotlin/Any?> kotlinx.coroutines.internal/kotlinx_coroutines_internal_JsIteratorResult_Companion_kfjag7_copy_1tks5(noinline kotlinx.coroutines.internal/JsIteratorResult<#A>, noinline #A? = ..., noinline kotlin/Boolean = ...): kotlinx.coroutines.internal/JsIteratorResult<#A> // kotlinx.coroutines.internal/kotlinx_coroutines_internal_JsIteratorResult_Companion_kfjag7_copy_1tks5|kotlinx_coroutines_internal_JsIteratorResult_Companion_kfjag7_copy_1tks5(kotlinx.coroutines.internal.JsIteratorResult<0:0>;0:0?;kotlin.Boolean){0§}[0] diff --git a/kotlinx-coroutines-core/concurrent/src/flow/Flow.concurrent.kt b/kotlinx-coroutines-core/concurrent/src/flow/Flow.concurrent.kt index a8c3ccb92f..b70e2647fd 100644 --- a/kotlinx-coroutines-core/concurrent/src/flow/Flow.concurrent.kt +++ b/kotlinx-coroutines-core/concurrent/src/flow/Flow.concurrent.kt @@ -1,195 +1,5 @@ package kotlinx.coroutines.flow -import kotlinx.coroutines.* -import kotlinx.coroutines.flow.internal.* -import kotlin.coroutines.* - -/** - * An asynchronous data stream that sequentially emits values and completes normally or with an exception. - * - * _Intermediate operators_ on the flow such as [map], [filter], [take], [zip], etc are functions that are - * applied to the _upstream_ flow or flows and return a _downstream_ flow where further operators can be applied to. - * Intermediate operations do not execute any code in the flow and are not suspending functions themselves. - * They only set up a chain of operations for future execution and quickly return. - * This is known as a _cold flow_ property. - * - * _Terminal operators_ on the flow are either suspending functions such as [collect], [single], [reduce], [toList], etc. - * or [launchIn] operator that starts collection of the flow in the given scope. - * They are applied to the upstream flow and trigger execution of all operations. - * Execution of the flow is also called _collecting the flow_ and is always performed in a suspending manner - * without actual blocking. Terminal operators complete normally or exceptionally depending on successful or failed - * execution of all the flow operations in the upstream. The most basic terminal operator is [collect], for example: - * - * ``` - * try { - * flow.collect { value -> - * println("Received $value") - * } - * } catch (e: Exception) { - * println("The flow has thrown an exception: $e") - * } - * ``` - * - * By default, flows are _sequential_ and all flow operations are executed sequentially in the same coroutine, - * with an exception for a few operations specifically designed to introduce concurrency into flow - * execution such as [buffer] and [flatMapMerge]. See their documentation for details. - * - * The `Flow` interface does not carry information whether a flow is a _cold_ stream that can be collected repeatedly and - * triggers execution of the same code every time it is collected, or if it is a _hot_ stream that emits different - * values from the same running source on each collection. Usually flows represent _cold_ streams, but - * there is a [SharedFlow] subtype that represents _hot_ streams. In addition to that, any flow can be turned - * into a _hot_ one by the [stateIn] and [shareIn] operators, or by converting the flow into a hot channel - * via the [produceIn] operator. - * - * ### Flow builders - * - * There are the following basic ways to create a flow: - * - * - [flowOf(...)][flowOf] functions to create a flow from a fixed set of values. - * - [asFlow()][asFlow] extension functions on various types to convert them into flows. - * - [flow { ... }][flow] builder function to construct arbitrary flows from - * sequential calls to [emit][FlowCollector.emit] function. - * - [channelFlow { ... }][channelFlow] builder function to construct arbitrary flows from - * potentially concurrent calls to the [send][kotlinx.coroutines.channels.SendChannel.send] function. - * - [MutableStateFlow] and [MutableSharedFlow] define the corresponding constructor functions to create - * a _hot_ flow that can be directly updated. - * - * ### Flow constraints - * - * All implementations of the `Flow` interface must adhere to two key properties described in detail below: - * - * - Context preservation. - * - Exception transparency. - * - * These properties ensure the ability to perform local reasoning about the code with flows and modularize the code - * in such a way that upstream flow emitters can be developed separately from downstream flow collectors. - * A user of a flow does not need to be aware of implementation details of the upstream flows it uses. - * - * ### Context preservation - * - * The flow has a context preservation property: it encapsulates its own execution context and never propagates or leaks - * it downstream, thus making reasoning about the execution context of particular transformations or terminal - * operations trivial. - * - * There is only one way to change the context of a flow: the [flowOn][Flow.flowOn] operator - * that changes the upstream context ("everything above the `flowOn` operator"). - * For additional information refer to its documentation. - * - * This reasoning can be demonstrated in practice: - * - * ``` - * val flowA = flowOf(1, 2, 3) - * .map { it + 1 } // Will be executed in ctxA - * .flowOn(ctxA) // Changes the upstream context: flowOf and map - * - * // Now we have a context-preserving flow: it is executed somewhere but this information is encapsulated in the flow itself - * - * val filtered = flowA // ctxA is encapsulated in flowA - * .filter { it == 3 } // Pure operator without a context yet - * - * withContext(Dispatchers.Main) { - * // All non-encapsulated operators will be executed in Main: filter and single - * val result = filtered.single() - * myUi.text = result - * } - * ``` - * - * From the implementation point of view, it means that all flow implementations should - * only emit from the same coroutine context. - * This constraint is efficiently enforced by the default [flow] builder. - * The [flow] builder should be used if the flow implementation does not start any coroutines. - * Its implementation prevents most of the development mistakes: - * - * ``` - * val myFlow = flow { - * // GlobalScope.launch { // is prohibited - * // launch(Dispatchers.IO) { // is prohibited - * // withContext(CoroutineName("myFlow")) { // is prohibited - * emit(1) // OK - * coroutineScope { - * emit(2) // OK -- still the same coroutine - * } - * } - * ``` - * - * Use [channelFlow] if the collection and emission of a flow are to be separated into multiple coroutines. - * It encapsulates all the context preservation work and allows you to focus on your - * domain-specific problem, rather than invariant implementation details. - * It is possible to use any combination of coroutine builders from within [channelFlow]. - * - * If you are looking for performance and are sure that no concurrent emits and context jumps will happen, - * the [flow] builder can be used alongside a [coroutineScope] or [supervisorScope] instead: - * - Scoped primitive should be used to provide a [CoroutineScope]. - * - Changing the context of emission is prohibited, no matter whether it is `withContext(ctx)` or - * a builder argument (e.g. `launch(ctx)`). - * - Collecting another flow from a separate context is allowed, but it has the same effect as - * applying the [flowOn] operator to that flow, which is more efficient. - * - * ### Exception transparency - * - * When `emit` or `emitAll` throws, the Flow implementations must immediately stop emitting new values and finish with an exception. - * For diagnostics or application-specific purposes, the exception may be different from the one thrown by the emit operation, - * suppressing the original exception as discussed below. - * If there is a need to emit values after the downstream failed, please use the [catch][Flow.catch] operator. - * - * The [catch][Flow.catch] operator only catches upstream exceptions, but passes - * all downstream exceptions. Similarly, terminal operators like [collect][Flow.collect] - * throw any unhandled exceptions that occur in their code or in upstream flows, for example: - * - * ``` - * flow { emitData() } - * .map { computeOne(it) } - * .catch { ... } // catches exceptions in emitData and computeOne - * .map { computeTwo(it) } - * .collect { process(it) } // throws exceptions from process and computeTwo - * ``` - * The same reasoning can be applied to the [onCompletion] operator that is a declarative replacement for the `finally` block. - * - * All exception-handling Flow operators follow the principle of exception suppression: - * - * If the upstream flow throws an exception during its completion when the downstream exception has been thrown, - * the downstream exception becomes superseded and suppressed by the upstream exception, being a semantic - * equivalent of throwing from `finally` block. However, this doesn't affect the operation of the exception-handling operators, - * which consider the downstream exception to be the root cause and behave as if the upstream didn't throw anything. - * - * Failure to adhere to the exception transparency requirement can lead to strange behaviors which make - * it hard to reason about the code because an exception in the `collect { ... }` could be somehow "caught" - * by an upstream flow, limiting the ability of local reasoning about the code. - * - * Flow machinery enforces exception transparency at runtime and throws [IllegalStateException] on any attempt to emit a value, - * if an exception has been thrown on previous attempt. - * - * ### Reactive streams - * - * 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. - * - * ### Not stable for inheritance - * - * **The `Flow` interface is not stable for inheritance in 3rd party libraries**, as new methods - * might be added to this interface in the future, but is stable for use. - * - * Use the `flow { ... }` builder function to create an implementation, or extend [AbstractFlow]. - * These implementations ensure that the context preservation property is not violated, and prevent most - * of the developer mistakes related to concurrency, inconsistent flow dispatchers, and cancellation. - */ public actual interface Flow { - - /** - * Accepts the given [collector] and [emits][FlowCollector.emit] values into it. - * - * This method can be used along with SAM-conversion of [FlowCollector]: - * ``` - * myFlow.collect { value -> println("Collected $value") } - * ``` - * - * ### Method inheritance - * - * To ensure the context preservation property, it is not recommended implementing this method directly. - * Instead, [AbstractFlow] can be used as the base type to properly ensure flow's properties. - * - * All default flow implementations ensure context preservation and exception transparency properties on a best-effort basis - * and throw [IllegalStateException] if a violation was detected. - */ public actual suspend fun collect(collector: FlowCollector) } diff --git a/kotlinx-coroutines-core/js/src/flow/Flow.js.kt b/kotlinx-coroutines-core/js/src/flow/Flow.js.kt index bf015a3716..125c046d61 100644 --- a/kotlinx-coroutines-core/js/src/flow/Flow.js.kt +++ b/kotlinx-coroutines-core/js/src/flow/Flow.js.kt @@ -10,198 +10,41 @@ import kotlinx.coroutines.internal.JsOptionalExport import kotlin.coroutines.EmptyCoroutineContext import kotlin.js.Promise -/** - * An asynchronous data stream that sequentially emits values and completes normally or with an exception. - * - * _Intermediate operators_ on the flow such as [map], [filter], [take], [zip], etc are functions that are - * applied to the _upstream_ flow or flows and return a _downstream_ flow where further operators can be applied to. - * Intermediate operations do not execute any code in the flow and are not suspending functions themselves. - * They only set up a chain of operations for future execution and quickly return. - * This is known as a _cold flow_ property. - * - * _Terminal operators_ on the flow are either suspending functions such as [collect], [single], [reduce], [toList], etc. - * or [launchIn] operator that starts collection of the flow in the given scope. - * They are applied to the upstream flow and trigger execution of all operations. - * Execution of the flow is also called _collecting the flow_ and is always performed in a suspending manner - * without actual blocking. Terminal operators complete normally or exceptionally depending on successful or failed - * execution of all the flow operations in the upstream. The most basic terminal operator is [collect], for example: - * - * ``` - * try { - * flow.collect { value -> - * println("Received $value") - * } - * } catch (e: Exception) { - * println("The flow has thrown an exception: $e") - * } - * ``` - * - * By default, flows are _sequential_ and all flow operations are executed sequentially in the same coroutine, - * with an exception for a few operations specifically designed to introduce concurrency into flow - * execution such as [buffer] and [flatMapMerge]. See their documentation for details. - * - * The `Flow` interface does not carry information whether a flow is a _cold_ stream that can be collected repeatedly and - * triggers execution of the same code every time it is collected, or if it is a _hot_ stream that emits different - * values from the same running source on each collection. Usually flows represent _cold_ streams, but - * there is a [SharedFlow] subtype that represents _hot_ streams. In addition to that, any flow can be turned - * into a _hot_ one by the [stateIn] and [shareIn] operators, or by converting the flow into a hot channel - * via the [produceIn] operator. - * - * ### Flow builders - * - * There are the following basic ways to create a flow: - * - * - [flowOf(...)][flowOf] functions to create a flow from a fixed set of values. - * - [asFlow()][asFlow] extension functions on various types to convert them into flows. - * - [flow { ... }][flow] builder function to construct arbitrary flows from - * sequential calls to [emit][FlowCollector.emit] function. - * - [channelFlow { ... }][channelFlow] builder function to construct arbitrary flows from - * potentially concurrent calls to the [send][kotlinx.coroutines.channels.SendChannel.send] function. - * - [MutableStateFlow] and [MutableSharedFlow] define the corresponding constructor functions to create - * a _hot_ flow that can be directly updated. - * - * ### Flow constraints - * - * All implementations of the `Flow` interface must adhere to two key properties described in detail below: - * - * - Context preservation. - * - Exception transparency. - * - * These properties ensure the ability to perform local reasoning about the code with flows and modularize the code - * in such a way that upstream flow emitters can be developed separately from downstream flow collectors. - * A user of a flow does not need to be aware of implementation details of the upstream flows it uses. - * - * ### Context preservation - * - * The flow has a context preservation property: it encapsulates its own execution context and never propagates or leaks - * it downstream, thus making reasoning about the execution context of particular transformations or terminal - * operations trivial. - * - * There is only one way to change the context of a flow: the [flowOn][Flow.flowOn] operator - * that changes the upstream context ("everything above the `flowOn` operator"). - * For additional information refer to its documentation. - * - * This reasoning can be demonstrated in practice: - * - * ``` - * val flowA = flowOf(1, 2, 3) - * .map { it + 1 } // Will be executed in ctxA - * .flowOn(ctxA) // Changes the upstream context: flowOf and map - * - * // Now we have a context-preserving flow: it is executed somewhere but this information is encapsulated in the flow itself - * - * val filtered = flowA // ctxA is encapsulated in flowA - * .filter { it == 3 } // Pure operator without a context yet - * - * withContext(Dispatchers.Main) { - * // All non-encapsulated operators will be executed in Main: filter and single - * val result = filtered.single() - * myUi.text = result - * } - * ``` - * - * From the implementation point of view, it means that all flow implementations should - * only emit from the same coroutine context. - * This constraint is efficiently enforced by the default [flow] builder. - * The [flow] builder should be used if the flow implementation does not start any coroutines. - * Its implementation prevents most of the development mistakes: - * - * ``` - * val myFlow = flow { - * // GlobalScope.launch { // is prohibited - * // launch(Dispatchers.IO) { // is prohibited - * // withContext(CoroutineName("myFlow")) { // is prohibited - * emit(1) // OK - * coroutineScope { - * emit(2) // OK -- still the same coroutine - * } - * } - * ``` - * - * Use [channelFlow] if the collection and emission of a flow are to be separated into multiple coroutines. - * It encapsulates all the context preservation work and allows you to focus on your - * domain-specific problem, rather than invariant implementation details. - * It is possible to use any combination of coroutine builders from within [channelFlow]. - * - * If you are looking for performance and are sure that no concurrent emits and context jumps will happen, - * the [flow] builder can be used alongside a [coroutineScope] or [supervisorScope] instead: - * - Scoped primitive should be used to provide a [CoroutineScope]. - * - Changing the context of emission is prohibited, no matter whether it is `withContext(ctx)` or - * a builder argument (e.g. `launch(ctx)`). - * - Collecting another flow from a separate context is allowed, but it has the same effect as - * applying the [flowOn] operator to that flow, which is more efficient. - * - * ### Exception transparency - * - * When `emit` or `emitAll` throws, the Flow implementations must immediately stop emitting new values and finish with an exception. - * For diagnostics or application-specific purposes, the exception may be different from the one thrown by the emit operation, - * suppressing the original exception as discussed below. - * If there is a need to emit values after the downstream failed, please use the [catch][Flow.catch] operator. - * - * The [catch][Flow.catch] operator only catches upstream exceptions, but passes - * all downstream exceptions. Similarly, terminal operators like [collect][Flow.collect] - * throw any unhandled exceptions that occur in their code or in upstream flows, for example: - * - * ``` - * flow { emitData() } - * .map { computeOne(it) } - * .catch { ... } // catches exceptions in emitData and computeOne - * .map { computeTwo(it) } - * .collect { process(it) } // throws exceptions from process and computeTwo - * ``` - * The same reasoning can be applied to the [onCompletion] operator that is a declarative replacement for the `finally` block. - * - * All exception-handling Flow operators follow the principle of exception suppression: - * - * If the upstream flow throws an exception during its completion when the downstream exception has been thrown, - * the downstream exception becomes superseded and suppressed by the upstream exception, being a semantic - * equivalent of throwing from `finally` block. However, this doesn't affect the operation of the exception-handling operators, - * which consider the downstream exception to be the root cause and behave as if the upstream didn't throw anything. - * - * Failure to adhere to the exception transparency requirement can lead to strange behaviors which make - * it hard to reason about the code because an exception in the `collect { ... }` could be somehow "caught" - * by an upstream flow, limiting the ability of local reasoning about the code. - * - * Flow machinery enforces exception transparency at runtime and throws [IllegalStateException] on any attempt to emit a value, - * if an exception has been thrown on previous attempt. - * - * ### Reactive streams - * - * 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. - * - * ### Not stable for inheritance - * - * **The `Flow` interface is not stable for inheritance in 3rd party libraries**, as new methods - * might be added to this interface in the future, but is stable for use. - * - * Use the `flow { ... }` builder function to create an implementation, or extend [AbstractFlow]. - * These implementations ensure that the context preservation property is not violated, and prevent most - * of the developer mistakes related to concurrency, inconsistent flow dispatchers, and cancellation. - */ @JsOptionalExport(couldBeConvertedToExplicitExport = true) public actual interface Flow { + @JsExport.Ignore + public actual suspend fun collect(collector: FlowCollector) /** - * Accepts the given [collector] and [emits][FlowCollector.emit] values into it. + * Represents [Flow] as a JavaScript [AsyncIterable](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Iteration_protocols#the_async_iterator_and_async_iterable_protocols) + * + * This function is a shorthand for: + * `buffer(0).produceIn(GlobalScope)`. + * + * Use it when a [Flow] needs to be exposed to JavaScript APIs that consume + * `AsyncIterable` (for example, via `for await (...)`). + * + * The returned iterable is backed by a coroutine started in [GlobalScope], so its lifecycle + * is not bound to a structured coroutine scope. With `buffer(0)`, elements are relayed with + * rendezvous-style backpressure (producer and consumer synchronize per element). * - * This method can be used along with SAM-conversion of [FlowCollector]: + * Kotlin usage: * ``` - * myFlow.collect { value -> println("Collected $value") } + * val flow = flowOf(1, 2, 3) + * val asyncIterable = flow.asAsyncIterable() + * // pass asyncIterable to JS code expecting AsyncIterable * ``` * - * ### Method inheritance + * JavaScript/TypeScript usage: + * ```javascript + * for await (const value of flow.asAsyncIterable()) { + * console.log(value) + * } + *``` * - * To ensure the context preservation property, it is not recommended implementing this method directly. - * Instead, [AbstractFlow] can be used as the base type to properly ensure flow's properties. - * - * All default flow implementations ensure context preservation and exception transparency properties on a best-effort basis - * and throw [IllegalStateException] if a violation was detected. + * This API is experimental: behavior and lifecycle semantics may change in future releases. */ - @JsExport.Ignore - public actual suspend fun collect(collector: FlowCollector) - - + @ExperimentalCoroutinesApi public fun asAsyncIterable(): JsAsyncIterable = buffer(0).produceIn(GlobalScope) @@ -220,6 +63,7 @@ public actual interface Flow { * to properly clean up the async iterable. */ @JsStatic + @ExperimentalCoroutinesApi public fun from(async: JsAsyncIterable): Flow = from(async.asyncIterator()) @@ -232,6 +76,7 @@ public actual interface Flow { */ @JsStatic @JsName("fromAsyncGenerator") + @ExperimentalCoroutinesApi public fun from(generator: () -> JsAsyncIterator): Flow = flow { var completed = false val iterator = generator() @@ -260,6 +105,7 @@ public actual interface Flow { */ @JsStatic @JsName("fromAsyncIterator") + @ExperimentalCoroutinesApi public fun from(iterator: JsAsyncIterator): Flow = from { iterator } } diff --git a/kotlinx-coroutines-core/js/test/FlowInteropTest.kt b/kotlinx-coroutines-core/js/test/FlowInteropTest.kt index 0160b2dfac..0625361153 100644 --- a/kotlinx-coroutines-core/js/test/FlowInteropTest.kt +++ b/kotlinx-coroutines-core/js/test/FlowInteropTest.kt @@ -111,6 +111,7 @@ class FlowInteropTest : TestBase() { assertNextStepToBe(iterator, 2, done = false) val returnResult = iterator.`return`(42).await() assertEquals(true, returnResult.done) + assertEquals(42, returnResult.value) assertNextStepToBe(iterator, done = true) assertNextStepToBe(iterator, done = true) } @@ -147,9 +148,13 @@ class FlowInteropTest : TestBase() { @Test fun testAsyncIteratorToFlowCancellation() = runTest { var returnCalled = false + var lastIndex = -1 val asyncIterator = createAsyncIteratorWithCleanup( listOf(1, 2, 3, 4, 5), - onReturn = { returnCalled = true } + onReturn = { + returnCalled = true + lastIndex = it + } ) val flow = Flow.from(asyncIterator) val results = mutableListOf() @@ -157,6 +162,7 @@ class FlowInteropTest : TestBase() { assertEquals(listOf(1, 2), results) yield() // Allow cleanup to happen assertTrue(returnCalled, "return() should be called on cancellation") + assertEquals(2, lastIndex, "Not only 1 and 2 were requested by asyncIterator") } @Test @@ -261,7 +267,7 @@ class FlowInteropTest : TestBase() { return iterator } - private fun createAsyncIteratorWithCleanup(values: List, onReturn: () -> Unit): JsAsyncIterator { + private fun createAsyncIteratorWithCleanup(values: List, onReturn: (Int) -> Unit): JsAsyncIterator { var index = 0 val iterator = js("({})") iterator.next = fun(): Promise> { @@ -273,7 +279,7 @@ class FlowInteropTest : TestBase() { } } iterator.`return` = fun(): Promise> { - onReturn() + onReturn(index) return Promise.resolve(js("({ value: undefined, done: true })")) } iterator.`throw` = fun(error: Throwable): Promise> { diff --git a/kotlinx-coroutines-core/wasmJs/src/flow/Flow.wasm.kt b/kotlinx-coroutines-core/wasmJs/src/flow/Flow.wasm.kt index a8c3ccb92f..b70e2647fd 100644 --- a/kotlinx-coroutines-core/wasmJs/src/flow/Flow.wasm.kt +++ b/kotlinx-coroutines-core/wasmJs/src/flow/Flow.wasm.kt @@ -1,195 +1,5 @@ package kotlinx.coroutines.flow -import kotlinx.coroutines.* -import kotlinx.coroutines.flow.internal.* -import kotlin.coroutines.* - -/** - * An asynchronous data stream that sequentially emits values and completes normally or with an exception. - * - * _Intermediate operators_ on the flow such as [map], [filter], [take], [zip], etc are functions that are - * applied to the _upstream_ flow or flows and return a _downstream_ flow where further operators can be applied to. - * Intermediate operations do not execute any code in the flow and are not suspending functions themselves. - * They only set up a chain of operations for future execution and quickly return. - * This is known as a _cold flow_ property. - * - * _Terminal operators_ on the flow are either suspending functions such as [collect], [single], [reduce], [toList], etc. - * or [launchIn] operator that starts collection of the flow in the given scope. - * They are applied to the upstream flow and trigger execution of all operations. - * Execution of the flow is also called _collecting the flow_ and is always performed in a suspending manner - * without actual blocking. Terminal operators complete normally or exceptionally depending on successful or failed - * execution of all the flow operations in the upstream. The most basic terminal operator is [collect], for example: - * - * ``` - * try { - * flow.collect { value -> - * println("Received $value") - * } - * } catch (e: Exception) { - * println("The flow has thrown an exception: $e") - * } - * ``` - * - * By default, flows are _sequential_ and all flow operations are executed sequentially in the same coroutine, - * with an exception for a few operations specifically designed to introduce concurrency into flow - * execution such as [buffer] and [flatMapMerge]. See their documentation for details. - * - * The `Flow` interface does not carry information whether a flow is a _cold_ stream that can be collected repeatedly and - * triggers execution of the same code every time it is collected, or if it is a _hot_ stream that emits different - * values from the same running source on each collection. Usually flows represent _cold_ streams, but - * there is a [SharedFlow] subtype that represents _hot_ streams. In addition to that, any flow can be turned - * into a _hot_ one by the [stateIn] and [shareIn] operators, or by converting the flow into a hot channel - * via the [produceIn] operator. - * - * ### Flow builders - * - * There are the following basic ways to create a flow: - * - * - [flowOf(...)][flowOf] functions to create a flow from a fixed set of values. - * - [asFlow()][asFlow] extension functions on various types to convert them into flows. - * - [flow { ... }][flow] builder function to construct arbitrary flows from - * sequential calls to [emit][FlowCollector.emit] function. - * - [channelFlow { ... }][channelFlow] builder function to construct arbitrary flows from - * potentially concurrent calls to the [send][kotlinx.coroutines.channels.SendChannel.send] function. - * - [MutableStateFlow] and [MutableSharedFlow] define the corresponding constructor functions to create - * a _hot_ flow that can be directly updated. - * - * ### Flow constraints - * - * All implementations of the `Flow` interface must adhere to two key properties described in detail below: - * - * - Context preservation. - * - Exception transparency. - * - * These properties ensure the ability to perform local reasoning about the code with flows and modularize the code - * in such a way that upstream flow emitters can be developed separately from downstream flow collectors. - * A user of a flow does not need to be aware of implementation details of the upstream flows it uses. - * - * ### Context preservation - * - * The flow has a context preservation property: it encapsulates its own execution context and never propagates or leaks - * it downstream, thus making reasoning about the execution context of particular transformations or terminal - * operations trivial. - * - * There is only one way to change the context of a flow: the [flowOn][Flow.flowOn] operator - * that changes the upstream context ("everything above the `flowOn` operator"). - * For additional information refer to its documentation. - * - * This reasoning can be demonstrated in practice: - * - * ``` - * val flowA = flowOf(1, 2, 3) - * .map { it + 1 } // Will be executed in ctxA - * .flowOn(ctxA) // Changes the upstream context: flowOf and map - * - * // Now we have a context-preserving flow: it is executed somewhere but this information is encapsulated in the flow itself - * - * val filtered = flowA // ctxA is encapsulated in flowA - * .filter { it == 3 } // Pure operator without a context yet - * - * withContext(Dispatchers.Main) { - * // All non-encapsulated operators will be executed in Main: filter and single - * val result = filtered.single() - * myUi.text = result - * } - * ``` - * - * From the implementation point of view, it means that all flow implementations should - * only emit from the same coroutine context. - * This constraint is efficiently enforced by the default [flow] builder. - * The [flow] builder should be used if the flow implementation does not start any coroutines. - * Its implementation prevents most of the development mistakes: - * - * ``` - * val myFlow = flow { - * // GlobalScope.launch { // is prohibited - * // launch(Dispatchers.IO) { // is prohibited - * // withContext(CoroutineName("myFlow")) { // is prohibited - * emit(1) // OK - * coroutineScope { - * emit(2) // OK -- still the same coroutine - * } - * } - * ``` - * - * Use [channelFlow] if the collection and emission of a flow are to be separated into multiple coroutines. - * It encapsulates all the context preservation work and allows you to focus on your - * domain-specific problem, rather than invariant implementation details. - * It is possible to use any combination of coroutine builders from within [channelFlow]. - * - * If you are looking for performance and are sure that no concurrent emits and context jumps will happen, - * the [flow] builder can be used alongside a [coroutineScope] or [supervisorScope] instead: - * - Scoped primitive should be used to provide a [CoroutineScope]. - * - Changing the context of emission is prohibited, no matter whether it is `withContext(ctx)` or - * a builder argument (e.g. `launch(ctx)`). - * - Collecting another flow from a separate context is allowed, but it has the same effect as - * applying the [flowOn] operator to that flow, which is more efficient. - * - * ### Exception transparency - * - * When `emit` or `emitAll` throws, the Flow implementations must immediately stop emitting new values and finish with an exception. - * For diagnostics or application-specific purposes, the exception may be different from the one thrown by the emit operation, - * suppressing the original exception as discussed below. - * If there is a need to emit values after the downstream failed, please use the [catch][Flow.catch] operator. - * - * The [catch][Flow.catch] operator only catches upstream exceptions, but passes - * all downstream exceptions. Similarly, terminal operators like [collect][Flow.collect] - * throw any unhandled exceptions that occur in their code or in upstream flows, for example: - * - * ``` - * flow { emitData() } - * .map { computeOne(it) } - * .catch { ... } // catches exceptions in emitData and computeOne - * .map { computeTwo(it) } - * .collect { process(it) } // throws exceptions from process and computeTwo - * ``` - * The same reasoning can be applied to the [onCompletion] operator that is a declarative replacement for the `finally` block. - * - * All exception-handling Flow operators follow the principle of exception suppression: - * - * If the upstream flow throws an exception during its completion when the downstream exception has been thrown, - * the downstream exception becomes superseded and suppressed by the upstream exception, being a semantic - * equivalent of throwing from `finally` block. However, this doesn't affect the operation of the exception-handling operators, - * which consider the downstream exception to be the root cause and behave as if the upstream didn't throw anything. - * - * Failure to adhere to the exception transparency requirement can lead to strange behaviors which make - * it hard to reason about the code because an exception in the `collect { ... }` could be somehow "caught" - * by an upstream flow, limiting the ability of local reasoning about the code. - * - * Flow machinery enforces exception transparency at runtime and throws [IllegalStateException] on any attempt to emit a value, - * if an exception has been thrown on previous attempt. - * - * ### Reactive streams - * - * 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. - * - * ### Not stable for inheritance - * - * **The `Flow` interface is not stable for inheritance in 3rd party libraries**, as new methods - * might be added to this interface in the future, but is stable for use. - * - * Use the `flow { ... }` builder function to create an implementation, or extend [AbstractFlow]. - * These implementations ensure that the context preservation property is not violated, and prevent most - * of the developer mistakes related to concurrency, inconsistent flow dispatchers, and cancellation. - */ public actual interface Flow { - - /** - * Accepts the given [collector] and [emits][FlowCollector.emit] values into it. - * - * This method can be used along with SAM-conversion of [FlowCollector]: - * ``` - * myFlow.collect { value -> println("Collected $value") } - * ``` - * - * ### Method inheritance - * - * To ensure the context preservation property, it is not recommended implementing this method directly. - * Instead, [AbstractFlow] can be used as the base type to properly ensure flow's properties. - * - * All default flow implementations ensure context preservation and exception transparency properties on a best-effort basis - * and throw [IllegalStateException] if a violation was detected. - */ public actual suspend fun collect(collector: FlowCollector) } diff --git a/kotlinx-coroutines-core/wasmWasi/src/flow/Flow.wasm.kt b/kotlinx-coroutines-core/wasmWasi/src/flow/Flow.wasm.kt index a8c3ccb92f..b70e2647fd 100644 --- a/kotlinx-coroutines-core/wasmWasi/src/flow/Flow.wasm.kt +++ b/kotlinx-coroutines-core/wasmWasi/src/flow/Flow.wasm.kt @@ -1,195 +1,5 @@ package kotlinx.coroutines.flow -import kotlinx.coroutines.* -import kotlinx.coroutines.flow.internal.* -import kotlin.coroutines.* - -/** - * An asynchronous data stream that sequentially emits values and completes normally or with an exception. - * - * _Intermediate operators_ on the flow such as [map], [filter], [take], [zip], etc are functions that are - * applied to the _upstream_ flow or flows and return a _downstream_ flow where further operators can be applied to. - * Intermediate operations do not execute any code in the flow and are not suspending functions themselves. - * They only set up a chain of operations for future execution and quickly return. - * This is known as a _cold flow_ property. - * - * _Terminal operators_ on the flow are either suspending functions such as [collect], [single], [reduce], [toList], etc. - * or [launchIn] operator that starts collection of the flow in the given scope. - * They are applied to the upstream flow and trigger execution of all operations. - * Execution of the flow is also called _collecting the flow_ and is always performed in a suspending manner - * without actual blocking. Terminal operators complete normally or exceptionally depending on successful or failed - * execution of all the flow operations in the upstream. The most basic terminal operator is [collect], for example: - * - * ``` - * try { - * flow.collect { value -> - * println("Received $value") - * } - * } catch (e: Exception) { - * println("The flow has thrown an exception: $e") - * } - * ``` - * - * By default, flows are _sequential_ and all flow operations are executed sequentially in the same coroutine, - * with an exception for a few operations specifically designed to introduce concurrency into flow - * execution such as [buffer] and [flatMapMerge]. See their documentation for details. - * - * The `Flow` interface does not carry information whether a flow is a _cold_ stream that can be collected repeatedly and - * triggers execution of the same code every time it is collected, or if it is a _hot_ stream that emits different - * values from the same running source on each collection. Usually flows represent _cold_ streams, but - * there is a [SharedFlow] subtype that represents _hot_ streams. In addition to that, any flow can be turned - * into a _hot_ one by the [stateIn] and [shareIn] operators, or by converting the flow into a hot channel - * via the [produceIn] operator. - * - * ### Flow builders - * - * There are the following basic ways to create a flow: - * - * - [flowOf(...)][flowOf] functions to create a flow from a fixed set of values. - * - [asFlow()][asFlow] extension functions on various types to convert them into flows. - * - [flow { ... }][flow] builder function to construct arbitrary flows from - * sequential calls to [emit][FlowCollector.emit] function. - * - [channelFlow { ... }][channelFlow] builder function to construct arbitrary flows from - * potentially concurrent calls to the [send][kotlinx.coroutines.channels.SendChannel.send] function. - * - [MutableStateFlow] and [MutableSharedFlow] define the corresponding constructor functions to create - * a _hot_ flow that can be directly updated. - * - * ### Flow constraints - * - * All implementations of the `Flow` interface must adhere to two key properties described in detail below: - * - * - Context preservation. - * - Exception transparency. - * - * These properties ensure the ability to perform local reasoning about the code with flows and modularize the code - * in such a way that upstream flow emitters can be developed separately from downstream flow collectors. - * A user of a flow does not need to be aware of implementation details of the upstream flows it uses. - * - * ### Context preservation - * - * The flow has a context preservation property: it encapsulates its own execution context and never propagates or leaks - * it downstream, thus making reasoning about the execution context of particular transformations or terminal - * operations trivial. - * - * There is only one way to change the context of a flow: the [flowOn][Flow.flowOn] operator - * that changes the upstream context ("everything above the `flowOn` operator"). - * For additional information refer to its documentation. - * - * This reasoning can be demonstrated in practice: - * - * ``` - * val flowA = flowOf(1, 2, 3) - * .map { it + 1 } // Will be executed in ctxA - * .flowOn(ctxA) // Changes the upstream context: flowOf and map - * - * // Now we have a context-preserving flow: it is executed somewhere but this information is encapsulated in the flow itself - * - * val filtered = flowA // ctxA is encapsulated in flowA - * .filter { it == 3 } // Pure operator without a context yet - * - * withContext(Dispatchers.Main) { - * // All non-encapsulated operators will be executed in Main: filter and single - * val result = filtered.single() - * myUi.text = result - * } - * ``` - * - * From the implementation point of view, it means that all flow implementations should - * only emit from the same coroutine context. - * This constraint is efficiently enforced by the default [flow] builder. - * The [flow] builder should be used if the flow implementation does not start any coroutines. - * Its implementation prevents most of the development mistakes: - * - * ``` - * val myFlow = flow { - * // GlobalScope.launch { // is prohibited - * // launch(Dispatchers.IO) { // is prohibited - * // withContext(CoroutineName("myFlow")) { // is prohibited - * emit(1) // OK - * coroutineScope { - * emit(2) // OK -- still the same coroutine - * } - * } - * ``` - * - * Use [channelFlow] if the collection and emission of a flow are to be separated into multiple coroutines. - * It encapsulates all the context preservation work and allows you to focus on your - * domain-specific problem, rather than invariant implementation details. - * It is possible to use any combination of coroutine builders from within [channelFlow]. - * - * If you are looking for performance and are sure that no concurrent emits and context jumps will happen, - * the [flow] builder can be used alongside a [coroutineScope] or [supervisorScope] instead: - * - Scoped primitive should be used to provide a [CoroutineScope]. - * - Changing the context of emission is prohibited, no matter whether it is `withContext(ctx)` or - * a builder argument (e.g. `launch(ctx)`). - * - Collecting another flow from a separate context is allowed, but it has the same effect as - * applying the [flowOn] operator to that flow, which is more efficient. - * - * ### Exception transparency - * - * When `emit` or `emitAll` throws, the Flow implementations must immediately stop emitting new values and finish with an exception. - * For diagnostics or application-specific purposes, the exception may be different from the one thrown by the emit operation, - * suppressing the original exception as discussed below. - * If there is a need to emit values after the downstream failed, please use the [catch][Flow.catch] operator. - * - * The [catch][Flow.catch] operator only catches upstream exceptions, but passes - * all downstream exceptions. Similarly, terminal operators like [collect][Flow.collect] - * throw any unhandled exceptions that occur in their code or in upstream flows, for example: - * - * ``` - * flow { emitData() } - * .map { computeOne(it) } - * .catch { ... } // catches exceptions in emitData and computeOne - * .map { computeTwo(it) } - * .collect { process(it) } // throws exceptions from process and computeTwo - * ``` - * The same reasoning can be applied to the [onCompletion] operator that is a declarative replacement for the `finally` block. - * - * All exception-handling Flow operators follow the principle of exception suppression: - * - * If the upstream flow throws an exception during its completion when the downstream exception has been thrown, - * the downstream exception becomes superseded and suppressed by the upstream exception, being a semantic - * equivalent of throwing from `finally` block. However, this doesn't affect the operation of the exception-handling operators, - * which consider the downstream exception to be the root cause and behave as if the upstream didn't throw anything. - * - * Failure to adhere to the exception transparency requirement can lead to strange behaviors which make - * it hard to reason about the code because an exception in the `collect { ... }` could be somehow "caught" - * by an upstream flow, limiting the ability of local reasoning about the code. - * - * Flow machinery enforces exception transparency at runtime and throws [IllegalStateException] on any attempt to emit a value, - * if an exception has been thrown on previous attempt. - * - * ### Reactive streams - * - * 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. - * - * ### Not stable for inheritance - * - * **The `Flow` interface is not stable for inheritance in 3rd party libraries**, as new methods - * might be added to this interface in the future, but is stable for use. - * - * Use the `flow { ... }` builder function to create an implementation, or extend [AbstractFlow]. - * These implementations ensure that the context preservation property is not violated, and prevent most - * of the developer mistakes related to concurrency, inconsistent flow dispatchers, and cancellation. - */ public actual interface Flow { - - /** - * Accepts the given [collector] and [emits][FlowCollector.emit] values into it. - * - * This method can be used along with SAM-conversion of [FlowCollector]: - * ``` - * myFlow.collect { value -> println("Collected $value") } - * ``` - * - * ### Method inheritance - * - * To ensure the context preservation property, it is not recommended implementing this method directly. - * Instead, [AbstractFlow] can be used as the base type to properly ensure flow's properties. - * - * All default flow implementations ensure context preservation and exception transparency properties on a best-effort basis - * and throw [IllegalStateException] if a violation was detected. - */ public actual suspend fun collect(collector: FlowCollector) } From 7425a1135c3e95ba287af96a8e8491e4d19da2fb Mon Sep 17 00:00:00 2001 From: Artem Kobzar Date: Fri, 7 Aug 2026 19:08:56 +0200 Subject: [PATCH 5/5] fixup! [K/JS] Add exportability of Flow with a conversion method to `AsyncIterator` --- kotlinx-coroutines-core/js/test/FlowInteropTest.kt | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/kotlinx-coroutines-core/js/test/FlowInteropTest.kt b/kotlinx-coroutines-core/js/test/FlowInteropTest.kt index 0625361153..5885b437d1 100644 --- a/kotlinx-coroutines-core/js/test/FlowInteropTest.kt +++ b/kotlinx-coroutines-core/js/test/FlowInteropTest.kt @@ -173,11 +173,10 @@ class FlowInteropTest : TestBase() { ) val flow = Flow.from(asyncIterator) val results = mutableListOf() - try { + assertFailsWith { flow.collect { results.add(it) } - fail("Should have thrown TestException") - } catch (e: TestException) { - assertEquals("iterator error", e.message) + }.apply { + assertEquals("iterator error", message) } assertEquals(listOf(1, 2), results) }