Kotlinx.coroutines: zipWithNext for Flow

[ThanosFisherman]

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.

[RinOrz]

@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

[RinOrz]

@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