fun <T> Flow<T>.onChange(action: (oldValue: T, newValue: T) -> Unit): Flow<T>

runningFold

runningReduce

fun <T> Flow<T>.zipWithNext(): Flow<Pair<T, T>>
fun <T, R> Flow<T>.zipWithNext(action: suspend FlowCollector<R>.(oldValue: T, newValue: T) -> Unit): Flow<R>

distinctUntilChangedBy

fun <T> Flow<T>.distinctUntilChangedWith(predicate: (T, T) -> Boolean): Flow<T>

fun <T> Flow<T>.onChange(action: suspend (oldValue: T, newValue: T) -> Unit): Flow<T> {
    return runningReduce { old, new ->
        action(old, new)
        new
    }
}