[2.0.0] repeatWhen/retryWhen memory leak

[nsk-mironov]

The following test case fails with RxJava-2.0.0, but works fine with 1.2.1:

public class RepeatWhenTest {
  @Test
  public void shouldDisposeInnerObservable() {
    final PublishSubject<String> subject = PublishSubject.create();
    final Disposable disposable = Observable.just("Leak")
        .repeatWhen(completions -> completions.switchMap(ignore -> subject))
        .subscribe();

    assertTrue(subject.hasObservers());
    disposable.dispose();
    assertFalse(subject.hasObservers());
  }
}

[nsk-mironov]

The same issue with retryWhen:

public class RetryWhenTest {
  @Test
  public void shouldDisposeInnerObservable() {
    final PublishSubject<String> subject = PublishSubject.create();
    final Disposable disposable = Observable.error(new RuntimeException("Leak"))
        .retryWhen(errors -> errors.switchMap(ignore -> subject))
        .subscribe();

    assertTrue(subject.hasObservers());
    disposable.dispose();
    assertFalse(subject.hasObservers());
  }
}

[akarnokd]

Thanks. I'll investigate the reasons and post a fix. The Flowable variants seem to work.

[akarnokd]

See #4783

[akarnokd]

Closing via #4783