<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 org.junit.Test;
import rx.Observable;
import rx.Subscription;
import rx.subjects.PublishSubject;
import rx.subjects.ReplaySubject;

public class RxUtilsTest extends TestCase {

    @Test
        assertThat(observable.blockingFirst()).isEqualTo("initial");

        // 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);
        assertThat(observable.toBlocking().first()).isEqualTo("first");
    }

}

<MSG> More RxUtils tests

<DFF> @@ -6,9 +6,16 @@ import junit.framework.TestCase;
 import org.junit.Test;
 import rx.Observable;
 import rx.Subscription;
+import rx.functions.Action0;
+import rx.functions.Func0;
+import rx.functions.Func1;
 import rx.subjects.PublishSubject;
 import rx.subjects.ReplaySubject;
 
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
 public class RxUtilsTest extends TestCase {
 
     @Test
@@ -33,4 +40,94 @@ public class RxUtilsTest extends TestCase {
         assertThat(observable.toBlocking().first()).isEqualTo("first");
     }
 
+    @Test
+    public static void testFromNullable() {
+        final Observable<String> fromNull = RxUtils.fromNullable(null);
+        assertThat(fromNull.toBlocking().getIterator().hasNext()).isFalse();
+
+        final Observable<String> fromNonNull = RxUtils.fromNullable("foo");
+        assertThat(fromNonNull.toBlocking().single()).isEqualTo("foo");
+    }
+
+    @Test
+    public static void testDeferredNullable() {
+        final Observable<String> fromNull = RxUtils.deferredNullable(new Func0<String>() {
+            @Override
+            public String call() {
+                return null;
+            }
+        });
+        assertThat(fromNull.toBlocking().getIterator().hasNext()).isFalse();
+
+        final Observable<String> fromNonNull = RxUtils.deferredNullable(new Func0<String>() {
+            @Override
+            public String call() {
+                return "foo";
+            }
+        });
+        assertThat(fromNonNull.toBlocking().single()).isEqualTo("foo");
+    }
+
+    @Test
+    public static void testWaitForCompletion() {
+        final PublishSubject<String> observable = PublishSubject.create();
+        final AtomicBoolean terminated = new AtomicBoolean(false);
+        new Thread() {
+            @Override
+            public void run() {
+                RxUtils.waitForCompletion(observable.toBlocking());
+                terminated.set(true);
+            }
+        }.start();
+        observable.onNext("foo");
+        assertThat(terminated.get()).isFalse();
+        observable.onNext("bar");
+        assertThat(terminated.get()).isFalse();
+        observable.onCompleted();
+        try {
+            Thread.sleep(100);
+        } catch (final InterruptedException ignored) {
+        }
+        assertThat(terminated.get()).isTrue();
+    }
+
+    @Test
+    public static void testObservableCache() {
+        final AtomicInteger counter = new AtomicInteger(0);
+        final RxUtils.ObservableCache<String, Integer> cache = new RxUtils.ObservableCache<String, Integer>(new Func1<String, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> call(final String s) {
+                counter.incrementAndGet();
+                return Observable.just(s.length());
+            }
+        });
+        assertThat(cache.get("a").toBlocking().single()).isEqualTo(1);
+        assertThat(counter.get()).isEqualTo(1);
+        assertThat(cache.get("a").toBlocking().single()).isEqualTo(1);
+        assertThat(counter.get()).isEqualTo(1);
+        assertThat(cache.get("bb").toBlocking().single()).isEqualTo(2);
+        assertThat(counter.get()).isEqualTo(2);
+        assertThat(cache.get("bb").toBlocking().single()).isEqualTo(2);
+        assertThat(counter.get()).isEqualTo(2);
+        assertThat(cache.get("a").toBlocking().single()).isEqualTo(1);
+        assertThat(counter.get()).isEqualTo(2);
+    }
+
+    @Test
+    public static void testDelayedUnsubscription() {
+        final AtomicBoolean unsubscribed = new AtomicBoolean(false);
+        Observable.never().doOnUnsubscribe(new Action0() {
+            @Override
+            public void call() {
+                unsubscribed.set(true);
+            }
+        }).lift(new RxUtils.DelayedUnsubscription<Object>(100, TimeUnit.MILLISECONDS)).subscribe().unsubscribe();
+        assertThat(unsubscribed.get()).isFalse();
+        try {
+            Thread.sleep(200);
+        } catch (final InterruptedException ignored) {
+        }
+        assertThat(unsubscribed.get()).isTrue();
+    }
+
 }

More RxUtils tests