GroupBy, filter and memory leak in Rx

Question

According to the documentation groupBy:

Note. A GroupedObservablewill cache the items it should release until it is signed. For this reason, to avoid memory leaks, you should not just ignore those GroupedObservablethat you are not interested in. Instead, you can tell them that they can discard their buffers by applying a type operator to them take(int)(0).

There's a RxJava tutorial that says:

Inside each Rx statement does 3 things

• He subscribes to the source and monitors the values.
• It converts the observed sequence according to the purpose of the operator.
• It pushes the changed sequence to its subscribers by calling onNext, onError and onCompleted.

Let's take a look at the following code block, which extracts only even numbers from 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);

My questions:

• Does this mean that the operator filteralready implies a subscription to each group as a result, groupByor only Observable<GroupedObservable>one?

• Will there be a memory leak in this case? If yes,

• How to delete these groups? Replace filterwith a custom that runs take(0), followed by return Observable.empty()? You may ask why I do not just return take(0)directly: this is because it filterdoes not have to be immediately after groupBy, but it can be anywhere in the chain and include more complex conditions.

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, .