Changes.kt

import io.reactivex.Observable

/**
 * RxJavaのStreamに流れる一つ前の値と現在の値をひとまとめにするための型
 */
data class Change<T : Any>(
    val previous: T?,
    val current: T,
)

/**
 * Observableに流れてくる一つ前の値と現在の値をひとまとめにした型に変換するoperator
 */
fun <T : Any> Observable<T>.changes(): Observable<Change<T>> {
    return scan(emptyList()) { list: List<T>, value: T ->
        (list + value).takeLast(2)
    }.filter {
        it.isNotEmpty()
    }.map { values ->
        when (values.size) {
            1 -> Change(null, values[0])
            2 -> Change(values[0], values[1])
            else -> throw IllegalStateException()
        }
    }
}