take(n) не имеет эффекта после groupBy в RxJava2

Question

Я пытаюсь сгруппировать несколько экземпляров модели по имени, а затем использовать take (n), чтобы взять только определенные элементы в группу, но каким-то образом дубль не влияет на GroupedObservable, Вот код

Давайте предположим, что он содержит список из 10 элементов, 5 из которых имеют название "яблоко", а остальные 5 имеют название "груша".

Observable<Item> items....

Observable<Item> groupedItems = items.groupBy(Item::name)
.flatMap(it -> it.take(2));

Так что я предполагаю, что groupedItems должен испускать 2 "яблока" и 2 "груши", но вместо этого у них есть все.

Есть ли что-то, что я ошибаюсь, мне нужно сделать это по-другому?

Answer 1

Отмененные группы воссоздаются при повторном обнаружении того же ключа. Вы должны убедиться, что группа не остановлена, и вам придется каким-то образом игнорировать другие элементы:

source.groupBy(func)
.flatMap(group -> 
    group.publish(p -> p.take(5).mergeWith(p.ignoreElements()))
);