[K/JS] Add exportability for Kotlin/JS with conversion to `AsyncItera…

[JSMonk]

…tor`

With the following commit the Flow interface could be consumed on the JavaScript / TypeScript side as follows:

for await (const value of FLOW_PRODUCER) { ... }

As well as created from from either AsyncIterable, AsyncIterator or an async generator. With an async iterator the API looks like the following:

import { Flow } from "coroutines"

Flow.fromAsyncGenerator(async function* () {
  yield await fetch(...)
  yield "Hello, World"
});

^KT-80733 Fixed

[JSMonk]

@dkhalanskyjb.

Important note: it could work only starting 2.3.20 (since @JsSymbol is introduced since this version).
So it could be merged only after 2.3.20

[dkhalanskyjb]

We cannot accept this PR as is. We already have several integrations with other asynchronous systems (for example, everything in the reactive/ directory), and this PR is inconsistent with that approach.

• Most notably, we usually don't have Flow implement any interfaces (even if it could have)—instead, we provide converter functions, so that the concerns are clearly separated.
• from functions are questionable: only from(JsAsyncIterable) is in line with the cold Flow semantics. In any scenario, it's more consistent for from to be a extension function on JsAsyncIterable instead.
• Also, none of the new code is tested.

If you would like to work on getting this PR to be merged, please take a look at the existing reactive integrations for inspiration and try mimicking their API. Alternatively, please file an issue instead, and we'll look into implementing the integration ourselves, provided there's a demand for it.

[JSMonk]

@dkhalanskyjb the functions are not designed to be used from Kotlin, but from the TypeScript side (as a part of the @JsExport functionality):

• For the first item, there is no other way to do so (if we didn't have any Flow to be passable as AsyncIterable inside TypeScript code, however, since the structural behavior of the TypeScript type system, I can do it just by adding the Symbol but escaping the AsyncIterable inside the super-type list)
• I agree that the part of implementation is not cold, but I disagree with the extension function part. Extension functions are designed to be easily used in Kotlin, but not on the other platforms. Since the factory functions are designed to be used from TypeScript, it makes sense to keep them within the static scope.
• Sure, I will add them; however, it's still required to migrate at least to 2.3.20 (so the code could at least compile).

[JSMonk]

@dkhalanskyjb

• Tests are added
• Super type is dropped
• Generator based solution is re-worked with a cold approach

[dkhalanskyjb]

Philosophically, the primary issue with the pull request is that, as proposed, the collection of Flow in JS is entirely different from the collection in Kotlin. collect is using the caller's coroutine context, collection of asynchronous iterators isn't. collect is push-based, AsyncIterable is pull-based.

With this mismatch, it's simply incorrect to have the async iterator API as readily available as collect. In fact, the less idiomatic + more brittle approach using JsAsyncIterable looks more streamlined with the proposed change than collect! Even in Kotlin, we can't write for (element in flow).

This means that some wrapping ceremony is mandatory here, at least to signify the travel between the two worlds. There is a conceptual chasm between Flow and AsyncIterable, and it should be reflected in the API. The non-context-preserving option needs to be chosen by the user implicitly, not hidden in some magic syntax.

The most idiomatic way to implement the conversion, one that's consistent with the other reactive integrations, is via extension functions, like

fun <T> Flow<T>.asAsyncIterable(): JsAsyncIterable<T>
fun <T> JsAsyncIterable<T>.asFlow(): Flow<T>

We do this for integration with the Java entities as well: https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-jdk9/kotlinx.coroutines.jdk9/ (see asFlow and asPublisher). I don't remember any complaints about these functions being somehow unfit for the purpose. They do the job of converting between various asynchronous systems.

The syntax from the Java side is indeed unpleasant, as you mention: you have to write something like ReactiveFlowKt.asPublisher(flow). The way to get around that is to perform the conversion on the Kotlin side, and then expose a pure Java API to the Java code. Still, when you're in a tight spot and have to call the extension function from Java code, you can do that.

Another way would be to introduce a wrapper class:

public class FlowIterable<out T>(public val flow: Flow<T>): JsAsyncIterable

This can be called with the same amount of ceremony from Kotlin and JS. It's not as nice as extension functions, as we typically short-circuit chains like asFlow().asPublisher() into no-ops to minimize the mismatch, and you can't do that with class wrappers. A class is also more API-heavy than a simple function.

[JSMonk]

Philosophically, the primary issue with the pull request is that, as proposed, the collection of Flow in JS is entirely different from the collection in Kotlin. collect uses the caller's coroutine context; a collection of asynchronous iterators isn't. collect is push-based; AsyncIterable is pull-based.

Exactly the same situation we already have with the Flow conversion to AsyncIteratorProtocol in SwiftExport, and we already convert them in this way, so to not break consistency across platforms - it doesn't make sense to introduce manual convertors.

The core idea behind our work on interop (both JS and Swift) is to make the use of Kotlin declarations on the target platform idiomatic for that platform (not for Kotlin, in our case, JS).

Unfortunately, there is no primitive for push-based async iteration in JavaScript, so the only way to represent Flow in an idiomatic JS way is AsyncIterable.

Another way would be to introduce a wrapper class:

public class FlowIterable<out T>(public val flow: Flow<T>): JsAsyncIterable

This can be called with the same amount of ceremony from Kotlin and JS. It's not as nice as extension functions, as we typically short-circuit chains like asFlow().asPublisher() into no-ops to minimize the mismatch, and you can't do that with class wrappers. A class is also more API-heavy than a simple function.

It requires the user to explicitly call those wrappers (which can't be done from the common code, since the wrapper class would exist only for JS/WasmJS targets), which means tons of expect/actual declarations in the sake of (to be honest, I haven't gotten the point in the sake of what).

Just one more time to highlight the point: the introduced API (at least right now) is not for use inside of Kotlin (I can make it clear by adding a hidden deprecation), but TypeScript.

P.S. I think there could be a need to pass Flow to an imported JavaScript API that expects AsyncIterator/AsyncIterable, so in this case, we definitely will need to introduce a super type of AsyncIterable for Flow, but we can postpone this. At this point, I want to resolve the issue with exporting Flow.

[dkhalanskyjb]

we already have with the Flow conversion to AsyncIteratorProtocol in SwiftExport

Thank you, I hadn't realized this. This is probably a problem as well. Could you provide any links to this functionality (code, documentation, KEEPs)? I couldn't find any mentions of this implicit transformation or how it's performed.

make the use of Kotlin declarations on the target platform idiomatic for that platform (not for Kotlin, in our case, JS).

It requires the user to explicitly call those wrappers (which can't be done from the common code, since the wrapper class would exist only for JS/WasmJS targets), which means tons of expect/actual declarations in the sake of (to be honest, I haven't gotten the point in the sake of what).

Sorry, this point is unclear. If the whole goal is to use Flow from other languages, why is the common code a concern? Yes, FlowIterable (or the equivalent) only makes sense for JS, so what would be the point of a common-code declaration exposing it? The interface exposed to JS can include FlowIterable without affecting other targets.

in this case, we definitely will need to introduce a super type of AsyncIterable for Flow

If the functions intended for AsyncIterable are present in Flow at all—which is the entire contention here—then sure, AsyncIterable should also become the supertype, I can't imagine any problems with that.

in the sake of (to be honest, I haven't gotten the point in the sake of what).

It's for the sake of not breaking the basic contracts. From https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow/ (emphasis mine):

An asynchronous data stream that sequentially emits values and completes normally or with an exception.

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.

Providing a way to collect a Flow that breaks the basis of the conceptual model behind Flow with no extra ceremony, as the default, just to save an extension function invocation per Flow collection, doesn't look like a good compromise to me.

Of course, we are also the authors of the documentation and could change it, but having to add "(except when it's being collected on JS using the native async iterable syntax)" throughout the text suggests the abstractions are broken and that the proposed semantics isn't a good fit.

With JS, we are at least lucky enough to catch this early, before it reaches the users—I don't know what the situation with Swift is.

[JSMonk]

Thank you, I hadn't realized this. This is probably a problem as well. Could you provide any links to this functionality (code, documentation, KEEPs)? I couldn't find any mentions of this implicit transformation or how it's performed.

The code is placed here (all the .kt, .swift, .h files)

Sorry, this point is unclear. If the whole goal is to use Flow from other languages, why is the common code a concern?

Since I want to have a common logic exported to all the platforms. Imagine a project with native UI on platforms (SwiftUI, React, as an example).

I have a logic placed in commonMain that is shared across all the platforms. I definitely don't want to duplicate it (its main purpose is KMP in this case), so I just want to export it to the platform and use it there. So this is why the common code is a concern.

It's for the sake of not breaking the basic contracts. From https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow/ (emphasis mine):

Could you please demonstrate how usage of the AsyncIterator from JavaScript breaks the basic contract?

[dkhalanskyjb]

The code is placed here (all the .kt, .swift, .h files)

Thanks, I'll read it some time later.

I just want to export it to the platform and use it there

And by that, do you mean that you want to take a Kotlin module, export its bindings to another language without writing any extra glue code in Kotlin, and use it as is? This aspiration fails to work even for Java, to which Kotlin is the closest. Not just in terms of extension functions, which are used quite often, but also in value classes that perform unpredictable name mangling.

Could you please demonstrate how usage of the AsyncIterator from JavaScript breaks the basic contract?

Yes, of course.

flow {
    println("Emitting 10")
    emit(10)
    println("Done processing 10")
    delay(100)
    println("Emitting 2")
    emit(2)
    println("Done processing 2")
}.collect {
    println("Repeat $it times")
    repeat(it) {
        println("Processing $it...")
        delay(100)
    }
    println("Done!")
}

The Flow semantics describe the order of this code clearly: first, everything before emit(10) runs, then, the collect lambda runs to completion, then, Done processing 10 is printed. Unless asynchronous behavior is intentionally introduced by using a hot flow or a concurrency operator, this property is preserved, and it's quite important for flows.

The async iterator in a for loop does not and can not work like this: its body will be interleaved with the flow processing.

This got me thinking: it looks at a glance like AsyncIterable is similar to the Channel in kotlinx.coroutines—ChannelIterator is an example of iteration over a channel, similar to AsyncIterable, and ReceiveChannel.cancel can probably be used to emulate return/throw on the AsyncIterable. Isn't Channel a better candidate for the AsyncIterable interface? We already have ReceiveChannel.receiveAsFlow and Flow.produceIn—those could provide access to Flow.

[JSMonk]

flow {
    println("Emitting 10")
    emit(10)
    println("Done processing 10")
    delay(100)
    println("Emitting 2")
    emit(2)
    println("Done processing 2")
}.collect {
    println("Repeat $it times")
    repeat(it) {
        println("Processing $it...")
        delay(100)
    }
    println("Done!")
}

The Flow semantics describe the order of this code clearly: first, everything before emit(10) runs, then, the collect lambda runs to completion, then, Done processing 10 is printed. Unless asynchronous behavior is intentionally introduced by using a hot flow or a concurrency operator, this property is preserved, and it's quite important for flows.

The async iterator in a for loop does not and can not work like this: its body will be interleaved with the flow processing.

Thank you so much; With this example, is there any possibility to repeat the same semantic by using Flow on JavaScript (without AsyncIterator)?

This got me thinking: it looks at a glance like AsyncIterable is similar to the Channel in kotlinx.coroutines—ChannelIterator is an example of iteration over a channel, similar to AsyncIterable, and ReceiveChannel.cancel can probably be used to emulate return/throw on the AsyncIterable. Isn't Channel a better candidate for the AsyncIterable interface? We already have ReceiveChannel.receiveAsFlow and Flow.produceIn—those could provide access to Flow.

Unfortunately, no. This is not my idea; it's just what people ask us. I would be happy to collaborate to make it happen (to make Flow exportable to JavaScript in terms of JavaScript API)

And by that, do you mean that you want to take a Kotlin module, export its bindings to another language without writing any extra glue code in Kotlin, and use it as is? This aspiration fails even for Java, to which Kotlin is closest. Not just in terms of extension functions, which are used quite often, but also in value classes that perform unpredictable name mangling.

Let's move forward without spending too much time on this. The SwiftExport team and mine have been working on it for a few years, and we've already gotten good results (at least from user feedback).

[dkhalanskyjb]

Hi! The develop branch, which is where we're doing development (instead of master), now uses Kotlin 2.3.21.

[JSMonk]

As discussed with @dkhalanskyjb, the MR was re-worked based on the channel exportability and explicit method converting Flow into AsyncIterable.

Firstly, the #4632 should be merged, since the Flow implementation of asAsyncIterable is based on the Channel implementation of Symbol.asyncIterator.

[dkhalanskyjb]

The build doesn't pass. Updating the ABI dump should fix that.

[dkhalanskyjb]

This public function is missing a KDoc. In my opinion, in terms of describing the behaviors, it should be sufficient to specify that it's a shorthand for buffer(0).produceIn(GlobalScope) (that is, its implementation). A usage example and an explanation of when one needs to use this would be appropriate, too.

Also, this function should be either marked as @ExperimentalCoroutinesApi or hidden from Kotlin code. There are too many unclear aspects to it, and there is a potential for its behavior to change, so we can't call this a stable API.

[dkhalanskyjb]

Likewise, since we know in advance those functions may need to get moved somewhere else, they need to be marked as experimental.

[dkhalanskyjb]

actual definitions of Flow shouldn't include copies of the KDocs already present in common.

[dkhalanskyjb]

Why not cancel the async iterator if emit threw an exception?

Note: in this scenario, it is fine to treat CancellationExceptions as a normal return, because that's consistent with the enormous design problem affecting the entirety of kotlinx.coroutines: #3658 We will need to fix the issue in the entire library later.

[JSMonk]

But it cancels the iterator, isn't it?

if (!completed) {
  iterator.asDynamic().`return`().unsafeCast<Promise<*>>().await()                
}

[JSMonk]

Or do you mean to provide the exception thrown by emit to the iterator via the throw method?

[dkhalanskyjb]

Yep, exactly.

[dkhalanskyjb]

Please check that the flow receives the exception. For example (note: I haven't tried running this, mistakes are likely):

val deferred = CompletableDeferred<Throwable>()
val flow = flow {
    emit(1)
    try {
        emit(2)
    } catch (e: Throwable) {
        deferred.complete(e)
    }
}
// ...
// after failing the flow
assertSame(error, deferred.await())

[JSMonk]

The same question is here. We've discussed that throwing an error inside of the for await shouldn't change the state of Flow/Channel. And the suggestion is actually to check that it's changed. I feel that we should test the opposite, shouldn't we?

[dkhalanskyjb]

We've discussed that throwing an error inside of the for await shouldn't change the state of Flow/Channel.

Yes, and my position is still the same: the state of Flow/Channel should change: #4632 (comment) My point is that such behavior aligns with the intuition and expectations of JS programmers, and we have no reason to diverge from that.

I see now that during the review, I missed the commit that made return and throw only affect the channel, and because of that, I merged the version that doesn't change the state of Channel. It's my mistake, I'm sorry about that.

When it comes to Flow, there's no space for debate, the flow has to observe the exception. The reasoning here goes beyond the JS programmer expectations and concerns the core of what a Flow is. There are two types of Flow:

1. Cold flows, and
2. Hot flows.

Cold flow collection is semantically a self-contained computation, launched for each collector independently and cleaned up when a collector is no longer interested in receiving values.

flow {
  try {
    emit(1)
  } finally {
    delay(100.milliseconds)
    println("Done")
  }
}.first()
// will print `Done` and only return after that

What you're proposing is to drop the computation once the result is no longer needed—which leaks closeable resources and leaves the cleanup computations in finally block unrun. This isn't aligned with Flow semantics and needs to be fixed.

Hot flow collection is shared among collectors—but collectors fail independently. SharedFlow doesn't notice any failures in collectors, so it's perfectly safe to cancel collectors indiscriminately. Only the coroutine scope in which the collection is launched can react to failures.

[dkhalanskyjb]

Fixed the channel behavior in a separate PR: #4718 Again, my apologies for merging the previous version without noticing that the channel was no longer getting cancelled.

With that change, the proper cleanup for cold flows in this PR should happen automatically, so we'll be able to keep the straightforward buffer(0).produceIn implementation instead of conjuring up something completely separate.

[JSMonk]

Sorry, but I completely disagree with the behavior and statements.

1. It's un-debaggable (aka painful) default behavior. It will take hours for Kotlin users to find the break place in JavaScript where their Flow/Channel was canceled.

2. It's not "the intuition and expectations of JS programmers". There is no contract on the AsyncIterable level that it should be "closed" after its AsyncIterator break/return/throwing (only the iterator itself). Different AsyncIterables behave differently. ReadableStream is closing itself on break (but to protect users, it also guarantees privileged access to itself, so only one listener at a time can be), while EventEmmitter doesn't close itself, and all the other listeners continue getting new events.

3. We're on the same ground that the AsyncIterable should preserve the Flow contracts. And I definitely want to provide users with the control over such a behavior (for both Flow and Channel). My proposal is to have different implementations depending on the Flow, so that if it's the cold one, to cancel it on break/return/throw (shall we also provide the same guarantees for the privileged access as ReadableStream? I feel we could, but not sure to do it in this iteration), for the "hot" flow, to keep the original flow as is without cancelling it.

4. Regarding Channel, it's a question, because I don't know the contract (at least the only thing I see that early break on Kotlin side iterations doesn't cancel the ReceiveChannel, it works only in the opposite way: if the channel is closed, the next method will throw the ClosedReceiveChannelException). However, I still believe that the least painful behavior for the Channel is the contract, that ReceiveChannel can be canceled only by the cancel method (which, btw, is also available from the TypeScript) and not implicitly by the early return/break/throw.
   Also, there is another argument for keeping the Channel behavior as it is now: with the cancel method available and the current behavior on early return, users have more control and can choose whether to use cancel or not (ReceiveChannel).
   Changing the behavior will remove this control (so users will not be able to stop listening to the channel without canceling it).

5. From my side, I feel that the discussion is already too long in time, and we definitely will not cover all the user expectations and cases with the discussion, so let we annotate all of the new methods with the experimental annotations (so opt-in is required, keeping the room for changing the behavior in the future after user feedback) and release at least some version to receive actual user feedback.

[JSMonk]

• Ah, got it - this behaves as expected with the particular implementation. The CancellationException is raised only when entering the second for await.
• For Flow, it also looks good from my side that this aligns with collect semantics.
• I’m still concerned about the default channel closing on break without providing privileged access to it.
  Also, I realized asAsyncIterable feels unnecessary for Flow, since it creates a new Channel per call and offers no meaningful configuration surface there. For Channel, however, such configuration does make sense, but currently there’s no dedicated API to control that behavior explicitly.
• For usability, I’d like to ask again that we consider:
  • making Flow directly AsyncIterable by default (no wrapper call when there is nothing to configure, e.g. no scope/cancellation options), and
  • adding an explicit conversion/configuration function for Channel, where cancellation behavior is actually configurable (similar to ReadableStream behavior options).

[JSMonk]

So, my proposal here is the following:

• In Channel, add a method similar to values in ReadableStream with the option to not cancel the channel (and delegate creating of the iterator to the method with the cancellation option as true)
• Inside Flow iterator mimics the API of Channel: default iterator which delegates to the default Channel iterator, and values that delegate to the channel values.

I'm proposing this specific API to follow the same idea as the:

• ReadableStream in browser
• Stream API in Node.js
• ReadableStream in Deno
• Stream API from Bun

So the API and control over the semantics would be familiar to the JS devs

[dkhalanskyjb]

In general—deal, let's do that. Let's figure out how to deal with the remaining edge cases.

Flow

The reason we decided to go for an explicit function for Flow was not exactly that we wanted to parameterize it, but rather, that there's a conceptual mismatch between normal Flow collection and the equivalent to buffer(0).produceIn(GlobalScope).

1. Flow executes synchronously, and all cleanup is done by the time collect returns. Example: https://pl.kotl.in/y8pdr-kXf
2. Flow executes in the caller's coroutine scope, not in GlobalScope.

2. can be worked around by configuring the desired context directly in .flowOn, but 1) is unavoidable with this simple implementation.

Counter-proposal

We could write a custom AsyncIterable implementation for Flow that deals with the asynchronous cleanup, without delegating to Channel. Since return and break return promises, we can wait until collect finishes before resolving them with either the result or an error. This deals with the major remaining mismatch and makes Flow collection from JS seamless.

Channel

Having a separate configurable method sounds okay. I'm not sure about its exact API, though.

• If we're going to expose it to Kotlin, it should look Kotlin-esque, which means something like fun ReceiveChannel<T>.asAsyncIterable(cancelOnEarlyExit: Boolean): AsyncIterable<T> (probably with some default value for cancelOnEarlyExit).
• However, the ReadableStream API is nothing like that, at all. Its values accepts an object with a single preventCancel field. Maybe that's familiar to JS users (and should be implemented for them as the target audience), but class ChannelAsyncIteratorOptions(val cancelOnEarlyExit: Boolean) is extremely non-conventional.

That is to say, either we hide these conversions from Kotlin (which could be okay), or we make two APIs, or the JS developers will have to deal with the Kotlin-esque API.

[JSMonk]

I've got it for the Channel and will prototype soon.
Regarding Flow. Would we want to have the version without cleaning (the same as the one in the Channel with preventCancel parameter)? If so, does it make sense just to use buffer(0).produceIn(GlobalScope). for the method?

[dkhalanskyjb]

No, there's no reason not to await a Flow cleanup. The cleanup will run anyway, we're only controlling whether it can run concurrently with the code that comes after for await.

[dkhalanskyjb]

Shouldn't returnResult.value be 42? Why not check that?

[dkhalanskyjb]

Please also test the no-argument throw.

[dkhalanskyjb]

We need to verify that finally cleanups at the point of cancellation run (and in fact, that cancellation does happen).

[JSMonk]

Should it happen? As far as I remember, we've discussed that throwing an error inside of for await should have no impact on either the Channel state or the Flow state.

[dkhalanskyjb]

Replied in the comment thread above. In short: yes, it's mandatory for a correct implementation of Flow collection.

[dkhalanskyjb]

We also need to check that 3, 4, and 5 weren't requested from asyncIterator (as opposed to just getting dropped somewhere on the way). To that end, onReturn could have the type (Int) -> Unit, where Int is the index of the last requested element.