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

import static org.assertj.core.api.Assertions.assertThat;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import static org.assertj.core.api.Java6Assertions.assertThat;

public class RxUtilsTest {

import rx.functions.Action0;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.subjects.PublishSubject;

public class RxUtilsTest extends TestCase {

        // Check that if the observable is not subscribed, changes are not propagated (similar to not keeping the
        // inner disposable active).
        rawObservable.onNext("without subscribers");
        assertThat(observable.blockingFirst()).isEqualTo("initial");

        // 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();
    }

}
        assertThat(RxUtils.nullableSingleValue(errorSingle)).isNull();
    }

}

<MSG> Add mergePreferred operator to RxJava

<DFF> @@ -2,6 +2,8 @@ package cgeo.geocaching.utils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.util.LinkedList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -13,6 +15,8 @@ import rx.Subscription;
 import rx.functions.Action0;
 import rx.functions.Func0;
 import rx.functions.Func1;
+import rx.observers.TestSubscriber;
+import rx.schedulers.TestScheduler;
 import rx.subjects.PublishSubject;
 
 public class RxUtilsTest extends TestCase {
@@ -110,4 +114,77 @@ public class RxUtilsTest extends TestCase {
         assertThat(RxUtils.nullableSingleValue(errorSingle)).isNull();
     }
 
+    public static void testMergePreferredImmediate() {
+        final Observable<Integer> o1 = Observable.range(0, 5);
+        final Observable<Integer> o2 = Observable.range(5, 5);
+        // take(100) is here to add some backpressure
+        final List<Integer> merged = RxUtils.mergePreferred(o1, o2).take(100).toList().toBlocking().single();
+        final List<Integer> expected = new LinkedList<>();
+        for (int i = 0; i < 10; i++) {
+            expected.add(i);
+        }
+        assertThat(merged).isEqualTo(expected);
+    }
+
+    public static void testMergePreferredDelayed() {
+        final TestScheduler scheduler = new TestScheduler();
+        final Observable<Integer> o1 = Observable.range(2, 3).delay(100, TimeUnit.MILLISECONDS, scheduler);
+        final Observable<Integer> o2 = Observable.range(10, 5);
+        final Observable<Integer> merged = RxUtils.mergePreferred(o1, o2);
+        final TestSubscriber<Integer> subscriber = new TestSubscriber<>(0);
+        merged.subscribe(subscriber);
+        subscriber.assertNoValues();
+        subscriber.requestMore(1);
+        subscriber.assertValues(10);  // From non-preferred, preferred is not ready
+        subscriber.requestMore(1);
+        subscriber.assertValues(10, 11);  // From non-preferred, preferred is not ready
+        scheduler.advanceTimeBy(120, TimeUnit.MILLISECONDS);
+        subscriber.requestMore(100);
+        subscriber.assertValues(10, 11, 2, 3, 4, 12, 13, 14);  // From non-preferred (preferred has completed)
+        subscriber.assertCompleted();
+    }
+
+    public static void testMergePreferredError() {
+        {
+            final PublishSubject<Integer> o1 = PublishSubject.create();
+            final PublishSubject<Integer> o2 = PublishSubject.create();
+            final Observable<Integer> merged = RxUtils.mergePreferred(o1, o2);
+            final TestSubscriber<Integer> subscriber = new TestSubscriber<>(0);
+            merged.subscribe(subscriber);
+            o1.onNext(1);
+            o1.onNext(2);
+            o2.onNext(11);
+            o2.onNext(12);
+            o2.onNext(13);
+            subscriber.assertNoValues();
+            subscriber.requestMore(3);
+            subscriber.assertValues(1, 2, 11);
+            o1.onNext(3);
+            subscriber.requestMore(100);
+            subscriber.assertValues(1, 2, 11, 3, 12, 13);
+            o1.onError(new RuntimeException());
+            subscriber.assertError(RuntimeException.class);
+        }
+        {
+            final PublishSubject<Integer> o1 = PublishSubject.create();
+            final PublishSubject<Integer> o2 = PublishSubject.create();
+            final Observable<Integer> merged = RxUtils.mergePreferred(o1, o2);
+            final TestSubscriber<Integer> subscriber = new TestSubscriber<>(0);
+            merged.subscribe(subscriber);
+            o1.onNext(1);
+            o1.onNext(2);
+            o2.onNext(11);
+            o2.onNext(12);
+            o2.onNext(13);
+            subscriber.assertNoValues();
+            subscriber.requestMore(3);
+            subscriber.assertValues(1, 2, 11);
+            o1.onNext(3);
+            subscriber.requestMore(100);
+            subscriber.assertValues(1, 2, 11, 3, 12, 13);
+            o2.onError(new RuntimeException());
+            subscriber.assertError(RuntimeException.class);
+        }
+    }
+
 }

Add mergePreferred operator to RxJava