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

Question

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

Я пытаюсь сравнить две наблюдаемые. Оба состоят из объектов Server с идентификатором и адресом_сервера.

Первая наблюдаемая представляет текущие объекты сервера, находящиеся в базе данных. Observable1 = [{"id":1, "server_address": "10.1.1.1},{"id":2, "server_address": "10.2.2.2},{"id":3, "server_address": "10.3.3.3}]

Вторая наблюдаемая была создана из запроса REST от пользователя серверных объектов, которые они хотели бы «добавить» / «вставить» в базу данных. Observable2 = [{"id":4, "server_address": "10.1.1.1},{"id":5, "server_address": "10.4.4.4}]

Я только хочу добавить / вставить новые серверы, которые НЕ соответствуют существующему адресу server_address, который уже находится в БД. Например, если поле server_address какого-либо из элементов в Observable2 совпадает с server_address из любого из элементов в Observable1, я хочу отфильтровать их и просто оставить действительно новые / уникальные элементы server_address в Observable2.

В приведенном выше примере мне понадобится логика, которая отфильтровывает сервер с идентификатором «4», потому что элемент с таким адресом сервера уже существует в БД. Поэтому я хочу отфильтровать элемент № 4, но сохранить элемент № 5. Затем я вставлю элемент 5 в базу данных.

Вопрос в том, какую операцию я могу использовать в RxJava2 для этого? Я видел много вещей о zip или объединения, но я не совсем уверен, что это то, что я хочу здесь.

Answer 1

Предполагая, что ваши наблюдаемые генерируют список элементов, вы, вероятно, захотите выполнить операцию фильтрации для коллекции, а не для наблюдаемой. Есть несколько способов сделать это. Вы можете достичь этого с помощью простой карты 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

Answer 2

Есть способы заставить это работать чисто реактивным способом, но, учитывая вашу спецификацию, это, вероятно, более сложно, чем использование другого подхода: сначала получите все, что вы хотите удалить, а затем отфильтруйте Observable2, используя это.

Воспринимайте это как псевдокод, поскольку я не совсем уверен, что вы хотите делать (и на каком языке вы это делаете), но это должно как-то служить:

allServers = observable1
    .map(s -> s['address'])
    .toList()
    .blockingGet() // out of the reactive world for a bit
    .toSet()

observable2
    .filter(s -> !allServers.contains(s['address']))
    .forEach(s -> doWhateverYouWantWith(s))

Если вы действительно хотите сделать это на 100% реактивным, вам, вероятно, понадобится способ отличить то, из какого источника вы наблюдаете, объединить результаты, сгруппировать по адресу и затем отфильтровать.Опять же, в псевдокоде и при условии, что у вас есть Map s, представляющих каждый элемент:

observable1
    .map(s -> s + {'source': 'db'})
    .merge(observable2.map(s -> s + {'source': 'request'})
    .groupBy(s -> s['address'])
    .flatMap(group -> group.take(1))
    .filter(s -> s['source'] == 'request')
    .forEach(s -> doWhateverYouWantWith(s))

Это работает, потому что в каждой группе вы будете иметь:

• толькоэлемент 'db' (который будет отфильтрован)
• и элемент 'db' и элемент 'request' (который будет отфильтрован, потому что 'db' всегда стоит первым)
• толькоэлемент 'request' (который пройдет фильтр)