zipWithNext for Flow

[ThanosFisherman]

trafficstars

Can we have a zipWithNext operator for Flow? Could you help me implement one?

[elizarov]

Implementation is pretty straight-forward, but what's your use-case? Why do you need it?

fun <T, R> Flow<T>.zipWithNext(transform: (T, T) -> R): Flow<R> = flow {
    var prev: Any? = UNDEFINED
    collect { value ->
        if (prev !== UNDEFINED) emit(transform(prev as T, value))
        prev = value
    }
}

private object UNDEFINED

[ThanosFisherman]

Thanks for helping out @elizarov

Well is not something that can't be done differently. I'm new to Coroutines but I am building a bluetooth library on Android using Coroutines/Flow.

As such during the readout response, which btw happens byte-by-byte via Flow, I need to check whether a specific pair of Hex bytes (02 FF) is included in the response when eventually collecting those bytes.

But I came up with an alternative using buffers and accumulating the whole Bluetooth Response into a ByteArray before it's collected by Flow. So in that case collect gives me a ByteArray instead of Byte so I can easily search for those two pair of values

[elizarov]

Good that you've found a better way. zipWithNext is a bad way to do it and using flow for byte-by-byte transfer is not ideal either.

[fluidsonic]

I would've found this useful for diffing upstream values into downstream change events.

datasets // DataSet1… DataSet2… DataSet3… DataSet… DataSet… 
   .onStart { emit(DataSet.empty) } // DataSet.empty… DataSet1… DataSet2… DataSet3… DataSet… DataSet… 
   .zipWithNext { a, b -> b.changesSince(a) } // [Change1a, Change1b, Change1c]… [Change2a, Change2b]…
   .flatMapConcat { changes -> changes.asFlow() } // Change1a… Change1b… Change1c… Change2a… Change2b… 

[elizarov]

@fluidsonic Thanks for a use-case.

[chachako]

@elizarov Should it be added to core? I currently need to merge the values of each emission, or is there a better way?

[elizarov]

@RinOrz You can use this implementation for now -> https://github.com/Kotlin/kotlinx.coroutines/issues/1767#issuecomment-577158308

[chachako]

@RinOrz You can use this implementation for now -> #1767 (comment)

Yes, I am currently doing this, I mean it might be added to the coroutine core, maybe it is useful

[matejdro]

Another use case: differentiating between existing state and between state changes.

For example, we have StateFlow<Boolean> in our app that emits true if user is logged in and false if user is logged out. For some operation we want to receive an event when user has just logged in. Just collecting the flow normally and triggering on true value is not helpful, because value could be true from the start (user could be logged in at the app startup). However, using zipWithNext allows us to know when value has changed from false to true.

We are using a bit modified version of the above implementation, where transform is also called on the first emission (but first parameter is null), so we can also get the initial value.