Use RxJava for direction and geodata updates rather than own observers

8 files changed, 171 insertions, 230 deletions

diff --git a/main/src/cgeo/geocaching/AbstractPopupActivity.java b/main/src/cgeo/geocaching/AbstractPopupActivity.java
index 5f24030..38e37da 100644
--- a/main/src/cgeo/geocaching/AbstractPopupActivity.java
+++ b/main/src/cgeo/geocaching/AbstractPopupActivity.java
@@ -40,7 +40,7 @@ public abstract class AbstractPopupActivity extends AbstractActivity implements
     private final GeoDirHandler geoUpdate = new GeoDirHandler() {
 
         @Override
-        protected void updateGeoData(final IGeoData geo) {
+        public void updateGeoData(final IGeoData geo) {
             try {
                 if (geo.getCoords() != null && cache != null && cache.getCoords() != null) {
                     cacheDistance.setText(Units.getDistanceFromKilometers(geo.getCoords().distanceTo(cache.getCoords())));
diff --git a/main/src/cgeo/geocaching/CgeoApplication.java b/main/src/cgeo/geocaching/CgeoApplication.java
index 265e7eb..1cea5f1 100644
--- a/main/src/cgeo/geocaching/CgeoApplication.java
+++ b/main/src/cgeo/geocaching/CgeoApplication.java
@@ -1,9 +1,10 @@
 package cgeo.geocaching;
 
 import cgeo.geocaching.ui.dialog.Dialogs;
-import cgeo.geocaching.utils.IObserver;
 import cgeo.geocaching.utils.Log;
 
+import rx.Observable;
+
 import android.app.Activity;
 import android.app.Application;
 import android.app.ProgressDialog;
@@ -13,8 +14,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
 
 public class CgeoApplication extends Application {
 
-    private volatile GeoDataProvider geo;
-    private volatile DirectionProvider dir;
+    private volatile Observable<IGeoData> geo;
+    private volatile Observable<Float> dir;
     private boolean forceRelog = false; // c:geo needs to log into cache providers
     public boolean showLoginToast = true; //login toast shown just once.
     private boolean liveMapHintShown = false; // livemap hint has been shown
@@ -65,29 +66,11 @@ public class CgeoApplication extends Application {
         }.start();
     }
 
-    /**
-     * Register an observer to receive GeoData information.
-     * <br/>
-     * If there is a chance that no observers are registered before this
-     * method is called, it is necessary to call it from a task implementing
-     * a looper interface as the data provider will use listeners that
-     * require a looper thread to run.
-     *
-     * @param observer a geodata observer
-     */
-    public void addGeoObserver(final IObserver<? super IGeoData> observer) {
-        currentGeoObject().addObserver(observer);
-    }
-
-    public void deleteGeoObserver(final IObserver<? super IGeoData> observer) {
-        currentGeoObject().deleteObserver(observer);
-    }
-
-    private GeoDataProvider currentGeoObject() {
+    public Observable<IGeoData> currentGeoObject() {
         if (geo == null) {
             synchronized(this) {
                 if (geo == null) {
-                    geo = new GeoDataProvider(this);
+                    geo = GeoDataProvider.create(this);
                 }
             }
         }
@@ -95,22 +78,14 @@ public class CgeoApplication extends Application {
     }
 
     public IGeoData currentGeo() {
-        return currentGeoObject().getMemory();
-    }
-
-    public void addDirectionObserver(final IObserver<? super Float> observer) {
-        currentDirObject().addObserver(observer);
-    }
-
-    public void deleteDirectionObserver(final IObserver<? super Float> observer) {
-        currentDirObject().deleteObserver(observer);
+        return currentGeoObject().first().toBlockingObservable().single();
     }
 
-    private DirectionProvider currentDirObject() {
+    public Observable<Float> currentDirObject() {
         if (dir == null) {
             synchronized(this) {
                 if (dir == null) {
-                    dir = new DirectionProvider(this);
+                    dir = DirectionProvider.create(this);
                 }
             }
         }
@@ -118,7 +93,7 @@ public class CgeoApplication extends Application {
     }
 
     public Float currentDirection() {
-        return currentDirObject().getMemory();
+        return currentDirObject().first().toBlockingObservable().single();
     }
 
     public boolean isLiveMapHintShown() {
diff --git a/main/src/cgeo/geocaching/CompassActivity.java b/main/src/cgeo/geocaching/CompassActivity.java
index 8955afd..6fc2de1 100644
--- a/main/src/cgeo/geocaching/CompassActivity.java
+++ b/main/src/cgeo/geocaching/CompassActivity.java
@@ -150,12 +150,13 @@ public class CompassActivity extends AbstractActivity {
         final CgeoApplication app = CgeoApplication.getInstance();
         final IGeoData geo = app.currentGeo();
         if (geo != null) {
-            geoDirHandler.update(geo);
+            geoDirHandler.updateGeoData(geo);
         }
         final Float dir = app.currentDirection();
         if (dir != null) {
-            geoDirHandler.update(dir);
+            geoDirHandler.updateDirection(dir);
         }
+
     }
 
     @Override
diff --git a/main/src/cgeo/geocaching/DirectionProvider.java b/main/src/cgeo/geocaching/DirectionProvider.java
index ae58fed..cc44155 100644
--- a/main/src/cgeo/geocaching/DirectionProvider.java
+++ b/main/src/cgeo/geocaching/DirectionProvider.java
@@ -1,9 +1,13 @@
 package cgeo.geocaching;
 
 import cgeo.geocaching.compatibility.Compatibility;
-import cgeo.geocaching.utils.MemorySubject;
 
-import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+import rx.observables.ConnectableObservable;
+import rx.subjects.BehaviorSubject;
 
 import android.app.Activity;
 import android.content.Context;
@@ -12,58 +16,61 @@ import android.hardware.SensorEvent;
 import android.hardware.SensorEventListener;
 import android.hardware.SensorManager;
 
-public class DirectionProvider extends MemorySubject<Float> implements SensorEventListener {
+public class DirectionProvider implements OnSubscribeFunc<Float> {
 
     private final SensorManager sensorManager;
+    private final BehaviorSubject<Float> subject = BehaviorSubject.create(0.0f);
 
-    // Previous values signaled to observers to avoid re-sending the same value when the
-    // device doesn't change orientation. The orientation is usually given with a 1 degree
-    // precision by Android, so it is not uncommon to obtain exactly the same value several
-    // times.
-    private float previous = -1;
-
-    public DirectionProvider(final Context context) {
-        sensorManager = (SensorManager) context.getSystemService(Context.SENSOR_SERVICE);
-
+    static public Observable<Float> create(final Context context) {
+        return new DirectionProvider((SensorManager) context.getSystemService(Context.SENSOR_SERVICE)).worker.refCount();
     }
 
-    @Override
-    protected void onFirstObserver() {
-        @SuppressWarnings("deprecation")
-        // This will be removed when using a new location service. Until then, it is okay to be used.
-        final Sensor defaultSensor = sensorManager.getDefaultSensor(Sensor.TYPE_ORIENTATION);
-        sensorManager.registerListener(this, defaultSensor, SensorManager.SENSOR_DELAY_NORMAL);
+    private DirectionProvider(final SensorManager sensorManager) {
+        this.sensorManager = sensorManager;
     }
 
     @Override
-    protected void onLastObserver() {
-        sensorManager.unregisterListener(this);
+    public Subscription onSubscribe(final Observer<? super Float> observer) {
+        return subject.distinctUntilChanged().subscribe(observer);
     }
 
-    @Override
-    public void onAccuracyChanged(final Sensor sensor, int accuracy) {
-        /*
-         * There is a bug in Android, which apparently causes this method to be called every
-         * time the sensor _value_ changed, even if the _accuracy_ did not change. So logging
-         * this event leads to the log being flooded with multiple entries _per second_,
-         * which I experienced when running cgeo in a building (with GPS and network being
-         * unreliable).
-         *
-         * See for example https://code.google.com/p/android/issues/detail?id=14792
-         */
+    private final ConnectableObservable<Float> worker = new ConnectableObservable<Float>(this) {
+        @Override
+        public Subscription connect() {
+            @SuppressWarnings("deprecation")
+            // This will be removed when using a new location service. Until then, it is okay to be used.
+            final Sensor defaultSensor = sensorManager.getDefaultSensor(Sensor.TYPE_ORIENTATION);
+            final SensorEventListener listener = new SensorEventListener() {
+                @Override
+                public void onSensorChanged(final SensorEvent event) {
+                    subject.onNext(event.values[0]);
+                }
 
-        //Log.i(Settings.tag, "Compass' accuracy is low (" + accuracy + ")");
-    }
+                @Override
+                public void onAccuracyChanged(final Sensor sensor, final int accuracy) {
+                /*
+                * There is a bug in Android, which apparently causes this method to be called every
+                * time the sensor _value_ changed, even if the _accuracy_ did not change. So logging
+                * this event leads to the log being flooded with multiple entries _per second_,
+                * which I experienced when running cgeo in a building (with GPS and network being
+                * unreliable).
+                *
+                * See for example https://code.google.com/p/android/issues/detail?id=14792
+                */
 
-    @Override
-    @SuppressFBWarnings("FE_FLOATING_POINT_EQUALITY")
-    public void onSensorChanged(final SensorEvent event) {
-        final float direction = event.values[0];
-        if (direction != previous) {
-            notifyObservers(direction);
-            previous = direction;
+                    //Log.i(Settings.tag, "Compass' accuracy is low (" + accuracy + ")");
+                }
+            };
+
+            sensorManager.registerListener(listener, defaultSensor, SensorManager.SENSOR_DELAY_NORMAL);
+            return new Subscription() {
+                @Override
+                public void unsubscribe() {
+                    sensorManager.unregisterListener(listener);
+                }
+            };
         }
-    }
+    };
 
     /**
      * Take the phone rotation (through a given activity) in account and adjust the direction.
@@ -72,6 +79,7 @@ public class DirectionProvider extends MemorySubject<Float> implements SensorEve
      * @param direction the unadjusted direction in degrees, in the [0, 360[ range
      * @return the adjusted direction in degrees, in the [0, 360[ range
      */
+
     public static float getDirectionNow(final Activity activity, final float direction) {
         return Compatibility.getDirectionNow(direction, activity);
     }
diff --git a/main/src/cgeo/geocaching/GeoDataProvider.java b/main/src/cgeo/geocaching/GeoDataProvider.java
index 73aefce..c61b7e7 100644
--- a/main/src/cgeo/geocaching/GeoDataProvider.java
+++ b/main/src/cgeo/geocaching/GeoDataProvider.java
@@ -3,9 +3,14 @@ package cgeo.geocaching;
 import cgeo.geocaching.enumerations.LocationProviderType;
 import cgeo.geocaching.geopoint.Geopoint;
 import cgeo.geocaching.utils.Log;
-import cgeo.geocaching.utils.MemorySubject;
 
 import org.apache.commons.lang3.StringUtils;
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+import rx.observables.ConnectableObservable;
+import rx.subjects.BehaviorSubject;
 
 import android.content.Context;
 import android.location.GpsSatellite;
@@ -15,22 +20,14 @@ import android.location.LocationListener;
 import android.location.LocationManager;
 import android.os.Bundle;
 
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.TimeUnit;
-
-/**
- * Provide information about the user location. This class should be instantiated only once per application.
- */
-class GeoDataProvider extends MemorySubject<IGeoData> {
+class GeoDataProvider implements OnSubscribeFunc<IGeoData> {
 
     private static final String LAST_LOCATION_PSEUDO_PROVIDER = "last";
     private final LocationManager geoManager;
-    private final GpsStatus.Listener gpsStatusListener = new GpsStatusListener();
     private final LocationData gpsLocation = new LocationData();
     private final LocationData netLocation = new LocationData();
-    private final Listener networkListener = new Listener(LocationManager.NETWORK_PROVIDER, netLocation);
-    private final Listener gpsListener = new Listener(LocationManager.GPS_PROVIDER, gpsLocation);
-    private final Unregisterer unregisterer = new Unregisterer();
+    private final BehaviorSubject<IGeoData> subject;
+
     public boolean gpsEnabled = false;
     public int satellitesVisible = 0;
     public int satellitesFixed = 0;
@@ -111,51 +108,6 @@ class GeoDataProvider extends MemorySubject<IGeoData> {
         }
     }
 
-    private class Unregisterer extends Thread {
-
-        private boolean unregisterRequested = false;
-        private final ArrayBlockingQueue<Boolean> queue = new ArrayBlockingQueue<Boolean>(1);
-
-        public void cancelUnregister() {
-            try {
-                queue.put(false);
-            } catch (final InterruptedException e) {
-                // Do nothing
-            }
-        }
-
-        public void lateUnregister() {
-            try {
-                queue.put(true);
-            } catch (final InterruptedException e) {
-                // Do nothing
-            }
-        }
-
-        @Override
-        public void run() {
-            try {
-                while (true) {
-                    if (unregisterRequested) {
-                        final Boolean element = queue.poll(2500, TimeUnit.MILLISECONDS);
-                        if (element == null) {
-                            // Timeout
-                            unregisterListeners();
-                            unregisterRequested = false;
-                        } else {
-                            unregisterRequested = element;
-                        }
-                    } else {
-                        unregisterRequested = queue.take();
-                    }
-                }
-            } catch (final InterruptedException e) {
-                // Do nothing
-            }
-        }
-
-    }
-
     /**
      * Build a new geo data provider object.
      * <p/>
@@ -164,10 +116,50 @@ class GeoDataProvider extends MemorySubject<IGeoData> {
      *
      * @param context the context used to retrieve the system services
      */
-    GeoDataProvider(final Context context) {
+    protected GeoDataProvider(final Context context) {
         geoManager = (LocationManager) context.getSystemService(Context.LOCATION_SERVICE);
-        unregisterer.start();
+        subject = BehaviorSubject.create(findInitialLocation());
+    }
+
+    public static Observable<IGeoData> create(final Context context) {
+        final GeoDataProvider provider = new GeoDataProvider(context);
+        return provider.worker.refCount();
+    }
 
+    @Override
+    public Subscription onSubscribe(final Observer<? super IGeoData> observer) {
+        return subject.subscribe(observer);
+    }
+
+    final ConnectableObservable<IGeoData> worker = new ConnectableObservable<IGeoData>(this) {
+        @Override
+        public Subscription connect() {
+            final GpsStatus.Listener gpsStatusListener = new GpsStatusListener();
+            geoManager.addGpsStatusListener(gpsStatusListener);
+
+            final Listener networkListener = new Listener(LocationManager.NETWORK_PROVIDER, netLocation);
+            final Listener gpsListener = new Listener(LocationManager.GPS_PROVIDER, gpsLocation);
+
+            for (final Listener listener : new Listener[] { networkListener, gpsListener }) {
+                try {
+                    geoManager.requestLocationUpdates(listener.locationProvider, 0, 0, listener);
+                } catch (final Exception e) {
+                    Log.w("There is no location provider " + listener.locationProvider);
+                }
+            }
+
+            return new Subscription() {
+                @Override
+                public void unsubscribe() {
+                    geoManager.removeUpdates(networkListener);
+                    geoManager.removeUpdates(gpsListener);
+                    geoManager.removeGpsStatusListener(gpsStatusListener);
+                }
+            };
+        }
+    };
+
+    private IGeoData findInitialLocation() {
         final Location initialLocation = new Location(LAST_LOCATION_PSEUDO_PROVIDER);
         try {
             // Try to find a sensible initial location from the last locations known to Android.
@@ -195,45 +187,13 @@ class GeoDataProvider extends MemorySubject<IGeoData> {
         }
         // Start with an historical GeoData just in case someone queries it before we get
         // a chance to get any information.
-        notifyObservers(new GeoData(initialLocation, false, 0, 0));
+        return new GeoData(initialLocation, false, 0, 0);
     }
 
     private static void copyCoords(final Location target, final Location source) {
         target.setLatitude(source.getLatitude());
         target.setLongitude(source.getLongitude());
     }
-    private void registerListeners() {
-        geoManager.addGpsStatusListener(gpsStatusListener);
-
-        for (final Listener listener : new Listener[] { networkListener, gpsListener }) {
-            try {
-                geoManager.requestLocationUpdates(listener.locationProvider, 0, 0, listener);
-            } catch (final Exception e) {
-                Log.w("There is no location provider " + listener.locationProvider);
-            }
-        }
-    }
-
-    private synchronized void unregisterListeners() {
-        // This method must be synchronized because it will be called asynchronously from the Unregisterer thread.
-        // We check that no observers have been re-added to prevent a race condition.
-        if (sizeObservers() == 0) {
-            geoManager.removeUpdates(networkListener);
-            geoManager.removeUpdates(gpsListener);
-            geoManager.removeGpsStatusListener(gpsStatusListener);
-        }
-    }
-
-    @Override
-    protected void onFirstObserver() {
-        unregisterer.cancelUnregister();
-        registerListeners();
-    }
-
-    @Override
-    protected void onLastObserver() {
-        unregisterer.lateUnregister();
-    }
 
     private class Listener implements LocationListener {
         private final String locationProvider;
@@ -336,7 +296,7 @@ class GeoDataProvider extends MemorySubject<IGeoData> {
         // We do not necessarily get signalled when satellites go to 0/0.
         final int visible = gpsLocation.isRecent() ? satellitesVisible : 0;
         final IGeoData current = new GeoData(locationData.location, gpsEnabled, visible, satellitesFixed);
-        notifyObservers(current);
+        subject.onNext(current);
     }
 
 }
diff --git a/main/src/cgeo/geocaching/maps/CGeoMap.java b/main/src/cgeo/geocaching/maps/CGeoMap.java
index c98ba72..a345d8d 100644
--- a/main/src/cgeo/geocaching/maps/CGeoMap.java
+++ b/main/src/cgeo/geocaching/maps/CGeoMap.java
@@ -898,7 +898,7 @@ public class CGeoMap extends AbstractMap implements OnMapDragListener, ViewFacto
         private long timeLastPositionOverlayCalculation = 0;
 
         @Override
-        protected void updateGeoData(final IGeoData geo) {
+        public void updateGeoData(final IGeoData geo) {
             if (geo.isPseudoLocation()) {
                 locationValid = false;
             } else {
diff --git a/main/src/cgeo/geocaching/speech/SpeechService.java b/main/src/cgeo/geocaching/speech/SpeechService.java
index 2a72bbf..8c650c3 100644
--- a/main/src/cgeo/geocaching/speech/SpeechService.java
+++ b/main/src/cgeo/geocaching/speech/SpeechService.java
@@ -47,7 +47,7 @@ public class SpeechService extends Service implements OnInitListener {
 
     GeoDirHandler geoHandler = new GeoDirHandler() {
         @Override
-        protected void updateDirection(float newDirection) {
+        public void updateDirection(float newDirection) {
             if (CgeoApplication.getInstance().currentGeo().getSpeed() <= 5) {
                 direction = DirectionProvider.getDirectionNow(startingActivity, newDirection);
                 directionInitialized = true;
@@ -56,7 +56,7 @@ public class SpeechService extends Service implements OnInitListener {
         }
 
         @Override
-        protected void updateGeoData(cgeo.geocaching.IGeoData newGeo) {
+        public void updateGeoData(cgeo.geocaching.IGeoData newGeo) {
             position = newGeo.getCoords();
             positionInitialized = true;
             if (!Settings.isUseCompass() || newGeo.getSpeed() > 5) {
diff --git a/main/src/cgeo/geocaching/utils/GeoDirHandler.java b/main/src/cgeo/geocaching/utils/GeoDirHandler.java
index c85648b..b9c605b 100644
--- a/main/src/cgeo/geocaching/utils/GeoDirHandler.java
+++ b/main/src/cgeo/geocaching/utils/GeoDirHandler.java
@@ -4,12 +4,15 @@ import cgeo.geocaching.CgeoApplication;
 import cgeo.geocaching.IGeoData;
 import cgeo.geocaching.settings.Settings;
 
-import android.os.Handler;
-import android.os.Message;
+import rx.Observable;
+import rx.Subscription;
+import rx.android.concurrency.AndroidSchedulers;
+import rx.util.functions.Action1;
+
+import java.util.concurrent.TimeUnit;
 
 /**
- * GeoData and Direction handler. Manipulating geodata and direction information
- * through a GeoDirHandler ensures that all listeners are registered from a {@link android.os.Looper} thread.
+ * GeoData and Direction handler.
  * <p>
  * To use this class, override at least one of {@link #updateDirection(float)} or {@link #updateGeoData(IGeoData)}. You
  * need to start the handler using one of
@@ -21,47 +24,11 @@ import android.os.Message;
  * A good place might be the {@code onResume} method of the Activity. Stop the Handler accordingly in {@code onPause}.
  * </p>
  */
-public abstract class GeoDirHandler extends Handler implements IObserver<Object> {
-
-    private static final int OBSERVABLE = 1 << 1;
-    private static final int START_GEO = 1 << 2;
-    private static final int START_DIR = 1 << 3;
-    private static final int STOP_GEO = 1 << 4;
-    private static final int STOP_DIR = 1 << 5;
-
+public abstract class GeoDirHandler {
     private static final CgeoApplication app = CgeoApplication.getInstance();
 
-    @Override
-    final public void handleMessage(final Message message) {
-        if ((message.what & START_GEO) != 0) {
-            app.addGeoObserver(this);
-        }
-
-        if ((message.what & START_DIR) != 0) {
-            app.addDirectionObserver(this);
-        }
-
-        if ((message.what & STOP_GEO) != 0) {
-            app.deleteGeoObserver(this);
-        }
-
-        if ((message.what & STOP_DIR) != 0) {
-            app.deleteDirectionObserver(this);
-        }
-
-        if ((message.what & OBSERVABLE) != 0) {
-            if (message.obj instanceof IGeoData) {
-                updateGeoData((IGeoData) message.obj);
-            } else {
-                updateDirection((Float) message.obj);
-            }
-        }
-    }
-
-    @Override
-    final public void update(final Object o) {
-        obtainMessage(OBSERVABLE, o).sendToTarget();
-    }
+    private Subscription dirSubscription = null;
+    private Subscription geoSubscription = null;
 
     /**
      * Update method called when new IGeoData is available.
@@ -69,7 +36,7 @@ public abstract class GeoDirHandler extends Handler implements IObserver<Object>
      * @param data
      *            the new data
      */
-    protected void updateGeoData(final IGeoData data) {
+    public void updateGeoData(final IGeoData data) {
         // Override this in children
     }
 
@@ -79,24 +46,38 @@ public abstract class GeoDirHandler extends Handler implements IObserver<Object>
      * @param direction
      *            the new direction
      */
-    protected void updateDirection(final float direction) {
+    public void updateDirection(final float direction) {
         // Override this in children
     }
 
     /**
      * Register the current GeoDirHandler for GeoData information.
      */
-    public void startGeo() {
-        sendEmptyMessage(START_GEO);
+    public synchronized void startGeo() {
+        geoSubscription = app.currentGeoObject()
+                .subscribeOn(AndroidSchedulers.mainThread())
+                .subscribe(new Action1<IGeoData>() {
+                    @Override
+                    public void call(final IGeoData geoData) {
+                        updateGeoData(geoData);
+                    }
+                });
     }
 
     /**
      * Register the current GeoDirHandler for direction information if the preferences
      * allow it.
      */
-    public void startDir() {
+    public synchronized void startDir() {
         if (Settings.isUseCompass()) {
-            sendEmptyMessage(START_DIR);
+            dirSubscription = app.currentDirObject()
+                    .subscribeOn(AndroidSchedulers.mainThread())
+                    .subscribe(new Action1<Float>() {
+                        @Override
+                        public void call(final Float direction) {
+                            updateDirection(direction);
+                        }
+                    });
         }
     }
 
@@ -105,27 +86,43 @@ public abstract class GeoDirHandler extends Handler implements IObserver<Object>
      * preferences allow it).
      */
     public void startGeoAndDir() {
-        sendEmptyMessage(START_GEO | (Settings.isUseCompass() ? START_DIR : 0));
+        startGeo();
+        startDir();
     }
 
     /**
      * Unregister the current GeoDirHandler for GeoData information.
      */
-    public void stopGeo() {
-        sendEmptyMessage(STOP_GEO);
+    public synchronized void stopGeo() {
+        // Delay the unsubscription by 2.5 seconds, so that another activity has
+        // the time to subscribe and the GPS receiver will not be turned down.
+        if (geoSubscription != null) {
+            final Subscription subscription = geoSubscription;
+            geoSubscription = null;
+            Observable.interval(2500, TimeUnit.MILLISECONDS).take(1).subscribe(new Action1<Long>() {
+                @Override
+                public void call(final Long aLong) {
+                    subscription.unsubscribe();
+                }
+            });
+        }
     }
 
     /**
      * Unregister the current GeoDirHandler for direction information.
      */
-    public void stopDir() {
-        sendEmptyMessage(STOP_DIR);
+    public synchronized void stopDir() {
+        if (dirSubscription != null) {
+            dirSubscription.unsubscribe();
+            dirSubscription = null;
+        }
     }
 
     /**
      * Unregister the current GeoDirHandler for GeoData and direction information.
      */
     public void stopGeoAndDir() {
-        sendEmptyMessage(STOP_GEO | STOP_DIR);
+        stopGeo();
+        stopDir();
     }
 }