Allows a stream to be filtered with a predicate that evaluates the latest and previous items.

rxFilterWithBiPredicate.kt

/**
 * Filters items emitted by an observable by only emitting those that satisfy the specified
 * predicate based on the latest and previous items emitted.
 * Note that this will always filter out the first item of the stream since in this case there
 * is as of yet no previous item to compare it to.
 *
 * @param bipredicate
 *            a function that evaluates the current and previous item emitted by the source ObservableSource, returning {@code true}
 *            if it passes the filter
 * @return an Observable that emits only those items emitted by the source ObservableSource that the filter
 *         evaluates as {@code true}
 */
fun <T> Observable<T>.filterWithBiPredicate(bipredicate: (T, T) -> Boolean): Observable<T> {
    return this.buffer(2, 1)
            .filter { buffer -> buffer.size == 2 }
            .filter { lastTwoItems ->
                val (old, current) = lastTwoItems
                bipredicate(old, current)
            }
            .map { lastTwoItems -> lastTwoItems[1] }
}