группеПо, фильтр и утечка памяти в Rx

Question

В соответствии с документацией groupBy:

Примечание: GroupedObservable будет кэшировать элементы это не испускать до тех пор, пока он подписан. По этой причине, чтобы избежать утечек памяти, вы не должны просто игнорировать те GroupedObservable, которые вас не интересуют. Вместо этого вы можете сообщить им, что они могут сбросить свои буферы, применив к ним оператора, например take(int)(0).

Там в RxJava tutorial, который говорит:

Внутренне, каждый оператор Rx делает 3 вещи

1. Он присоединяется к источнику и соблюдает значения.
2. Он преобразует наблюдаемую последовательность в соответствии с назначением оператора.
3. Он подталкивает измененную последовательность к своим подписчикам, вызывая onNext, onError и onCompleted.

Давайте посмотрим на следующий блок кода, который извлекает только четные числа от range(0, 10):

Observable.range(0, 10) 
     .groupBy(i -> i % 2) 
     .filter(g -> g.getKey() % 2 == 0) 
     .flatMap(g -> g) 
     .subscribe(System.out::println, Throwable::printStackTrace); 

Мои вопросы:

1. Означает ли это filter оператор уже подразумевает подписка на каждую группу была получена от groupBy или только от Observable<GroupedObservable>?

2. Будет ли утечка памяти в этом случае? Если это так,

3. Как правильно удалить эти группы? Замените filter на заказ, который выполняет take(0), а затем return Observable.empty()? Вы можете спросить, почему я не просто возвращаю take(0) напрямую: это потому, что filter не обязательно следует сразу после groupBy, но может быть где угодно в цепочке и включать более сложные условия.

Answer 1

Ваши подозрения верны в том, чтобы правильно обрабатывать Сгруппированная наблюдаемым каждой из внутренних наблюдаемых (g) должны быть подписаны. Поскольку filter подписывается на внешний наблюдаемый, это только плохая идея. Просто выполните то, что вам нужно в flatMap, используя ignoreElements для фильтрации нежелательных групп.

Observable.range(0, 10) 
    .groupBy(i -> i % 2) 
    .flatMap(g -> { 
     if (g.getKey() % 2 == 0) 
     return g; 
     else 
     return g.ignoreElements(); 
    }) 
    .subscribe(System.out::println, Throwable::printStackTrace); 

Answer 2

Помимо утечки памяти, текущая реализация может полностью зависать из-за внутренних проблем координации запросов.

Обратите внимание, что с использованием take(0) группа может быть воссоздана все время. Вместо этого я бы использовал ignoreElements, который понижает значения, а не доходит до flatMap, и сама группа не будет воссоздана все время.