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

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import rx.Subscriber;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.observables.BlockingObservable;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;
import rx.subscriptions.Subscriptions;
public class RxUtils {

    private RxUtils() {
        // Utility class, not to be instantiated
    }

    public static <T> Observable<T> rememberLast(final Observable<T> observable, final T initialValue) {
        final AtomicReference<T> lastValue = new AtomicReference<>(initialValue);
        return observable.doOnNext(lastValue::set).startWith(Observable.defer(() -> {
            final T last = lastValue.get();
            return last != null ? Observable.just(last) : Observable.empty();
        })).replay(1).refCount();
    }

    /**
     * Cache the last value of observables so that every key is associated to only one of them.
     *
     * @param <K> the type of the key
     * @param <V> the type of the value
     */
    public static class ObservableCache<K, V> {

        private final Function<K, Observable<V>> func;
        private final Map<K, Observable<V>> cached = new HashMap<>();

        /**
         * Create a new observables cache.
         *
         * @param func the function transforming a key into an observable
         */
        public ObservableCache(final Function<K, Observable<V>> func) {
            this.func = func;
        }

        /**
         * Get the observable corresponding to a key. If the key has not already been
         * seen, the function passed to the constructor will be called to build the observable
         * <p/>
         * If the observable has already emitted values, only the last one will be remembered.
         * <p/>
         * If the function throws an exception, it will be returned in the observable.
         *
         * @param key the key
         * @return the observable corresponding to the key
         */
        public synchronized Observable<V> get(final K key) {
            if (cached.containsKey(key)) {
                return cached.get(key);
            }
            try {
                final Observable<V> value = func.apply(key).replay(1).refCount();
                cached.put(key, value);
                return value;
            } catch (final Throwable t) {
                final Observable<V> error = Observable.error(t);
                cached.put(key, error);
                return error;
            }
        }


        @Override
        final public void call(final Subscriber<? super T> subscriber) {
            subscriber.add(subject.subscribe(new Action1<T>() {
                @Override
                public void call(final T data) {
                    subscriber.onNext(data);
                }
            }, new Action1<Throwable>() {
                @Override
                public void call(final Throwable throwable) {
                    subscriber.onError(throwable);
                }
            }, new Action0() {
                @Override
                public void call() {
                    subscriber.onCompleted();
                }
            }));
            looperCallbacksWorker.schedule(new Action0() {
                @Override
                public void call() {
        public DelayedUnsubscription(final long time, final TimeUnit unit) {
            this.time = time;
            this.unit = unit;
        }

        @Override
        public Observer<? super T> apply(final Observer<? super T> observer) throws Exception {
            final AtomicBoolean canceled = new AtomicBoolean();

            return new Observer<T>() {

                @Override
                public void onSubscribe(final Disposable d) {
                    observer.onSubscribe(new CancellableDisposable(() -> {
                        canceled.set(true);
                        Schedulers.computation().scheduleDirect(d::dispose, time, unit);
                    }));
                }

                @Override
                public void onComplete() {
                    if (!canceled.get()) {
                        observer.onComplete();
                    }
                }

                @Override
                public void onError(final Throwable e) {
                    if (!canceled.get()) {
                        observer.onError(e);
                    }
                }

                @Override
                public void onNext(final T t) {
                    if (!canceled.get()) {
                        observer.onNext(t);
                    }
                }
            };
        }
    }

}

<MSG> Use shorter form for cloned subscriber

<DFF> @@ -8,9 +8,9 @@ import rx.Scheduler.Worker;
 import rx.Subscriber;
 import rx.android.schedulers.AndroidSchedulers;
 import rx.functions.Action0;
-import rx.functions.Action1;
 import rx.functions.Func1;
 import rx.observables.BlockingObservable;
+import rx.observers.Subscribers;
 import rx.schedulers.Schedulers;
 import rx.subjects.PublishSubject;
 import rx.subscriptions.Subscriptions;
@@ -77,22 +77,7 @@ public class RxUtils {
 
         @Override
         final public void call(final Subscriber<? super T> subscriber) {
-            subscriber.add(subject.subscribe(new Action1<T>() {
-                @Override
-                public void call(final T data) {
-                    subscriber.onNext(data);
-                }
-            }, new Action1<Throwable>() {
-                @Override
-                public void call(final Throwable throwable) {
-                    subscriber.onError(throwable);
-                }
-            }, new Action0() {
-                @Override
-                public void call() {
-                    subscriber.onCompleted();
-                }
-            }));
+            subscriber.add(subject.subscribe(Subscribers.from(subscriber)));
             looperCallbacksWorker.schedule(new Action0() {
                 @Override
                 public void call() {

Use shorter form for cloned subscriber