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

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action0;
import rx.observables.BlockingObservable;
import rx.schedulers.Schedulers;
import rx.subscriptions.CompositeSubscription;
import rx.subscriptions.Subscriptions;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class RxUtils {

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

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

    private static final StartableHandlerThread looperCallbacksThread =
            new StartableHandlerThread("Looper callbacks thread", android.os.Process.THREAD_PRIORITY_BACKGROUND);

    static {
        looperCallbacksThread.start();
    }

    private static final Worker looperCallbacksWorker = AndroidSchedulers.handlerThread(looperCallbacksThread.getHandler()).createWorker();

    public final static Scheduler computationScheduler = Schedulers.computation();

    public static final Scheduler networkScheduler = Schedulers.from(new ThreadPoolExecutor(10, 10, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()));
    }

    /**
     * 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
    }

    /**
     * Start a job (typically one that register handlers) on a looper thread if the counter goes above 0,
     * and stop it if it comes back to 0 after the job is unsubscribed from.
     *
     * @param counter the counter to use
     * @param onStart the job to launch if the counter goes above 0
     * @param onStop the job to launch when unsubscribing if the counter goes back at 0
     * @param stopDelay the delay before which the unsubscription should take place
     * @param stopDelayUnit the unit of the delay before which the unsubscription should take place
     * @return the subscription allowing to unsubscribe
     */
    public static Subscription looperCallbacksSchedule(final AtomicInteger counter, final Action0 onStart, final Action0 onStop,
                                                       final long stopDelay, final TimeUnit stopDelayUnit) {
        final CompositeSubscription subscription = new CompositeSubscription();
        looperCallbacksWorker.schedule(new Action0() {
            @Override
            public void call() {
                if (counter.getAndIncrement() == 0) {
                    onStart.call();
                }
                subscription.add(Subscriptions.create(new Action0() {
                    @Override
                    public void call() {
                        looperCallbacksWorker.schedule(new Action0() {
                            @Override
                            public void call() {
                                if (counter.decrementAndGet() == 0) {
                                    onStop.call();
                                }
                            }
                        }, stopDelay, stopDelayUnit);
                    }
                }));
            }
        });
        return subscription;
    }

    /**
     * Start a job (typically one that register handlers) on a looper thread if the counter goes above 0,
     * and stop it if it comes back to 0 after the job is unsubscribed from.
     *
     * @param counter the counter to use
     * @param onStart the job to launch if the counter goes above 0
     * @param onStop the job to launch when unsubscribing if the counter goes back at 0
     * @return the subscription allowing to unsubscribe
     */
    public static Subscription looperCallbacksSchedule(final AtomicInteger counter, final Action0 onStart, final Action0 onStop) {
        return looperCallbacksSchedule(counter, onStart, onStop, 0, TimeUnit.SECONDS);
    }
}
                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> Further factoring of ConnectableObservable behaviour

<DFF> @@ -6,7 +6,9 @@ import rx.Scheduler.Worker;
 import rx.Subscription;
 import rx.android.schedulers.AndroidSchedulers;
 import rx.functions.Action0;
+import rx.functions.Action1;
 import rx.observables.BlockingObservable;
+import rx.observables.ConnectableObservable;
 import rx.schedulers.Schedulers;
 import rx.subscriptions.CompositeSubscription;
 import rx.subscriptions.Subscriptions;
@@ -21,15 +23,6 @@ public class RxUtils {
     // Utility class, not to be instanciated
     private RxUtils() {}
 
-    private static final StartableHandlerThread looperCallbacksThread =
-            new StartableHandlerThread("Looper callbacks thread", android.os.Process.THREAD_PRIORITY_BACKGROUND);
-
-    static {
-        looperCallbacksThread.start();
-    }
-
-    private static final Worker looperCallbacksWorker = AndroidSchedulers.handlerThread(looperCallbacksThread.getHandler()).createWorker();
-
     public final static Scheduler computationScheduler = Schedulers.computation();
 
     public static final Scheduler networkScheduler = Schedulers.from(new ThreadPoolExecutor(10, 10, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>()));
@@ -43,53 +36,60 @@ public class RxUtils {
     }
 
     /**
-     * Start a job (typically one that register handlers) on a looper thread if the counter goes above 0,
-     * and stop it if it comes back to 0 after the job is unsubscribed from.
+     * ConnectableObservable whose subscription and unsubscription take place on a looper thread.
      *
-     * @param counter the counter to use
-     * @param onStart the job to launch if the counter goes above 0
-     * @param onStop the job to launch when unsubscribing if the counter goes back at 0
-     * @param stopDelay the delay before which the unsubscription should take place
-     * @param stopDelayUnit the unit of the delay before which the unsubscription should take place
-     * @return the subscription allowing to unsubscribe
+     * @param <T> the type of the observable
      */
-    public static Subscription looperCallbacksSchedule(final AtomicInteger counter, final Action0 onStart, final Action0 onStop,
-                                                       final long stopDelay, final TimeUnit stopDelayUnit) {
-        final CompositeSubscription subscription = new CompositeSubscription();
-        looperCallbacksWorker.schedule(new Action0() {
-            @Override
-            public void call() {
-                if (counter.getAndIncrement() == 0) {
-                    onStart.call();
-                }
-                subscription.add(Subscriptions.create(new Action0() {
-                    @Override
-                    public void call() {
-                        looperCallbacksWorker.schedule(new Action0() {
-                            @Override
-                            public void call() {
-                                if (counter.decrementAndGet() == 0) {
-                                    onStop.call();
-                                }
-                            }
-                        }, stopDelay, stopDelayUnit);
+    public static abstract class ConnectableLooperCallbacks<T> extends ConnectableObservable<T> {
+        private static final StartableHandlerThread looperCallbacksThread =
+                new StartableHandlerThread("Looper callbacks thread", android.os.Process.THREAD_PRIORITY_BACKGROUND);
+        static {
+            looperCallbacksThread.start();
+        }
+        private static final Worker looperCallbacksWorker = AndroidSchedulers.handlerThread(looperCallbacksThread.getHandler()).createWorker();
+
+        final AtomicInteger counter = new AtomicInteger(0);
+        final long stopDelay;
+        final TimeUnit stopDelayUnit;
+
+        public ConnectableLooperCallbacks(final OnSubscribe<T> onSubscribe, final long stopDelay, final TimeUnit stopDelayUnit) {
+            super(onSubscribe);
+            this.stopDelay = stopDelay;
+            this.stopDelayUnit = stopDelayUnit;
+        }
+
+        public ConnectableLooperCallbacks(final OnSubscribe<T> onSubscribe) {
+            this(onSubscribe, 0, TimeUnit.SECONDS);
+        }
+
+        @Override
+        final public void connect(final Action1<? super Subscription> action1) {
+            final CompositeSubscription subscription = new CompositeSubscription();
+            looperCallbacksWorker.schedule(new Action0() {
+                @Override
+                public void call() {
+                    if (counter.getAndIncrement() == 0) {
+                        onStart();
                     }
-                }));
-            }
-        });
-        return subscription;
-    }
+                    subscription.add(Subscriptions.create(new Action0() {
+                        @Override
+                        public void call() {
+                            looperCallbacksWorker.schedule(new Action0() {
+                                @Override
+                                public void call() {
+                                    if (counter.decrementAndGet() == 0) {
+                                        onStop();
+                                    }
+                                }
+                            }, stopDelay, stopDelayUnit);
+                        }
+                    }));
+                }
+            });
+            action1.call(subscription);
+        }
 
-    /**
-     * Start a job (typically one that register handlers) on a looper thread if the counter goes above 0,
-     * and stop it if it comes back to 0 after the job is unsubscribed from.
-     *
-     * @param counter the counter to use
-     * @param onStart the job to launch if the counter goes above 0
-     * @param onStop the job to launch when unsubscribing if the counter goes back at 0
-     * @return the subscription allowing to unsubscribe
-     */
-    public static Subscription looperCallbacksSchedule(final AtomicInteger counter, final Action0 onStart, final Action0 onStop) {
-        return looperCallbacksSchedule(counter, onStart, onStop, 0, TimeUnit.SECONDS);
+        abstract protected void onStart();
+        abstract protected void onStop();
     }
 }

Further factoring of ConnectableObservable behaviour