Как сравнить элементы в двух наблюдаемых, а затем фильтровать на основе совпадений в RxJava2?

Предполагая, что ваши наблюдаемые генерируют список элементов, вы, вероятно, захотите выполнить операцию фильтрации для коллекции, а не для наблюдаемой. Есть несколько способов сделать это. Вы можете добиться этого с помощью простой карты flatMap + или с помощьюlateLatest / zip, как вы упомянули.

Примеры написаны на Kotlin.

val observable1 = Observable.just((1..10 step 2).toList()) // Odd integers between 1 and 10
val observable2 = Observable.just((1..10).toList()) // Integers between 1 and 10

Плоская карта + карта

observable1.flatMap { result1 ->
    observable2.map { result2 -> result2 - result1 }
}.subscribe { println(it.toString()) }

CombineLatest

Observable.combineLatest<List<Int>, List<Int>, List<Int>>(
    observable1,
    observable2,
    BiFunction { result1, result2 ->
        result2.filter { item ->
            !result1.contains(item)
        }
    })
    .subscribe { println(it.toString()) }

Zip

Observable.zip<List<Int>, List<Int>, List<Int>>(
    observable1,
    observable2,
    BiFunction { result1, result2 ->
        result2 - result1
    })
    .subscribe { println(it.toString()) }

Все эти выходные данные:

[2, 4, 6, 8, 10]

В качестве альтернативы, если вы хотите использовать оператор фильтра RxJava и генерировать каждый элемент отдельно, вы можете использовать flatMapIterable, чтобы преобразовать список в поток элементов и выполнить фильтр на каждый предмет.

observable1.flatMap { result1 ->
    observable2.flatMapIterable { result2 -> result2 }
        .filter { item -> !result1.contains(item) }
}.subscribe { println(it.toString()) }

Выход:

2
4
6
8
10