<NME> RxUtilsTest.java
<BEF> package cgeo.geocaching.utils;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import rx.Subscription;
import rx.functions.Func1;
import rx.subjects.PublishSubject;

import android.test.AndroidTestCase;

public class RxUtilsTest extends AndroidTestCase {

    public static void testTakeUntil() {
        final Observable<Integer> observable = Observable.range(1, 10).lift(RxUtils.operatorTakeUntil(new Func1<Integer, Boolean>() {
            @Override
            public Boolean call(final Integer value) {
                return value > 6;
        // Check that new values are propagated and cached
        final Disposable disposable = observable.subscribe();
        rawObservable.onNext("first");
        assertThat(observable.blockingFirst()).isEqualTo("first");
        disposable.dispose();
        assertThat(observable.blockingFirst()).isEqualTo("first");
    }

    @Test
    public void testObservableCache() {
        final AtomicInteger counter = new AtomicInteger(0);
        final RxUtils.ObservableCache<String, Integer> cache = new RxUtils.ObservableCache<>(s -> {
            counter.incrementAndGet();
            return Observable.just(s.length());
        });
        assertThat(cache.get("a").blockingSingle()).isEqualTo(1);
        assertThat(counter.get()).isEqualTo(1);
        assertThat(cache.get("a").blockingSingle()).isEqualTo(1);
        assertThat(counter.get()).isEqualTo(1);
        assertThat(cache.get("bb").blockingSingle()).isEqualTo(2);
        assertThat(counter.get()).isEqualTo(2);
        assertThat(cache.get("bb").blockingSingle()).isEqualTo(2);
        assertThat(counter.get()).isEqualTo(2);
        assertThat(cache.get("a").blockingSingle()).isEqualTo(1);
        assertThat(counter.get()).isEqualTo(2);
    }

    @Test
    public void testDelayedUnsubscription() {
        final AtomicBoolean unsubscribed = new AtomicBoolean(false);
        Observable.never().doOnDispose(() -> unsubscribed.set(true)).lift(new RxUtils.DelayedUnsubscription<>(100, TimeUnit.MILLISECONDS)).subscribe().dispose();
        assertThat(unsubscribed.get()).isFalse();
        try {
            Thread.sleep(200);
        } catch (final InterruptedException ignored) {
            // ignore for tests
        }
        assertThat(unsubscribed.get()).isTrue();
    }

}

<MSG> Cherry-picked and merged test fixes

<DFF> @@ -6,13 +6,23 @@ import rx.Observable;
 import rx.Subscription;
 import rx.functions.Func1;
 import rx.subjects.PublishSubject;
+import rx.subjects.ReplaySubject;
 
 import android.test.AndroidTestCase;
 
 public class RxUtilsTest extends AndroidTestCase {
 
+    // Observable.range(int, int) is not kept in the application by proguard. Use an explicit range here.
+    private static final ReplaySubject<Integer> range = ReplaySubject.createWithSize(10);
+    static {
+        for (int i = 1; i <= 10; i++) {
+            range.onNext(i);
+        }
+        range.onCompleted();
+    }
+
     public static void testTakeUntil() {
-        final Observable<Integer> observable = Observable.range(1, 10).lift(RxUtils.operatorTakeUntil(new Func1<Integer, Boolean>() {
+        final Observable<Integer> observable = range.lift(RxUtils.operatorTakeUntil(new Func1<Integer, Boolean>() {
             @Override
             public Boolean call(final Integer value) {
                 return value > 6;

Cherry-picked and merged test fixes