Hi! When I introduce a `scan` in my Flow chain, it starts to act strange — items emitted by channels...

# coroutines

[dimsuz]

Hi! When I introduce a scan in my Flow chain, it starts to act strange — items emitted by channels stop arriving into it. I tried to come up with minimal example to reproduce, posted on SO, would be grateful if anyone could take a look and help:

https://stackoverflow.com/q/63092171/258848

[Zach Klippenstein (he/him) [MOD]]

If you put an .onEach { println(it) } after extraValues.asFlow() when you pass it to merge, are you seeing the values there?

[dimsuz]

no, nothing there

[dimsuz]

If I remove scan, they start to be printed there too

[Zach Klippenstein (he/him) [MOD]]

I think I know what’s going on. Looking at the source, the seed accumulator value is emitted before starting to collect the upstream flow. Since you trigger the extraValues sends when that initial seed is emitted, those values will be sent into the channel before anyone is subscribed to it, which, per the BroadcastChannel documentation, means those values will be ignored (BroadcastChannels ignore values when there are no subscribers). Here’s the source: https://github.com/Kotlin/kotlinx.coroutines/blob/d55d8e80e152e46c170edc4dcfa08ccb7b2c4a03/kotlinx-coroutines-core/common/src/flow/operators/Transform.kt#L87

[Zach Klippenstein (he/him) [MOD]]

This kind of issue has come up with at least one other operator (I think it was onStart?), where the upstream collection was happening later than expected (after the operator’s lambda was invoked), and weird things happen. Or at least, they seem weird coming from maybe RxJava experience where the subscription signal is, I think, fully propagated upstream before anything happens downstream.

[dimsuz]

Exactly! onStart has this behavior too, I wanted to include this in my question, but decided not to add complexity 🙂

Thanks for explanation. I wonder how can I workaround this limitation. I really need the scan here. Not sure if it's possible to write some scan which would emit first value after collection has started

[Zach Klippenstein (he/him) [MOD]]

I would start by filing a bug for this and describing your use case in depth. I am curious if @elizarov has an idea how to fix this, and if he has any thoughts about this as a more general, recurring issue.

That said, it should be pretty straightforward to write your own version of the operator though, and you can fix the bug by subscribing upstream with a channel before emitting (i.e. with produceIn), and then just consuming the channel until it’s closed.

[julian]

This supports @Zach Klippenstein (he/him) [MOD]’s explanation: if
if (it == 1) is changed to if (it == 2), the code behaves as desired.

[dimsuz]

Yeah, it's not the first time I encounter this.

Basically I'm often using redux-like "loop", which scans through user produced actions and it can also produce a side-effect along with the state. this side effect can trigger some external source which is represented by the extraValues channel in my sample.

So whenever the very first reduce operation triggers this side effect, described behavior is observed: some incoming state is being lost...

[julian]

@Zach Klippenstein (he/him) [MOD] I'm having trouble understanding this

... subscribing upstream with a channel before emitting (i.e. with produceIn), and then just consuming the channel until it’s closed.

Can you explain a bit more?

[Zach Klippenstein (he/him) [MOD]]

coroutineScope {
  val subscription = [e-mail](this)
  emit(seed)
  subscription.consumeEach {
    // rest of scan logic
  }
}

[julian]

@Zach Klippenstein (he/him) [MOD] Ah, I think I see. This is a modification of the scan implementation?

[dimsuz]

by the way it seems like this has been discussed before, I have found this issue: https://github.com/Kotlin/kotlinx.coroutines/issues/1758

[Zach Klippenstein (he/him) [MOD]]

For the record, I don’t think MutableSharedFlow will fix this. I believe it has the same behavior as ArrayBroadcastChannel re: dropping items sent to it when there are no subscribers.

[dimsuz]

Hmm. The last posts in the issue above seem to indicate that SharedFlow has onSubscription operator designed specifically to support this case. Unless I've misread something.

[Zach Klippenstein (he/him) [MOD]]

That only works if you've got direct access to the SharedFlow though.

[dimsuz]

ah, yes. If this would be some API giving me only a Flow, no luck.