RXJava Location Observable

gistfile1.java

import android.content.Context;
import android.location.Criteria;
import android.location.Location;
import android.location.LocationListener;
import android.location.LocationManager;
import android.os.Bundle;
import android.os.Looper;

import rx.Observable;
import rx.Subscriber;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action0;
import rx.functions.Func1;
import rx.subscriptions.Subscriptions;
import timber.log.Timber;

public class LocationStreams {
    private static final long LOCATION_MIN_TIME_BETWEEN_UPDATES = 0L;
    private static final float LOCATION_MIN_DISTANCE_BETWEEN_UPDATES = 0f;

    private static final LocationManager LOCATION_MANAGER = (LocationManager) Application.CONTEXT.getSystemService(Context.LOCATION_SERVICE);

    private static final Observable<Location> mLocationObservable = createLocationObservable();

    private static Observable<Location> createLocationObservable() {
        return Observable.create(new Observable.OnSubscribe<Location>() {
            @Override public void call(final Subscriber<? super Location> subscriber) {
                Timber.d("Starting Location Services.");
                Criteria locationCriteria = new Criteria();
                locationCriteria.setSpeedAccuracy(Criteria.ACCURACY_HIGH);

                final LocationListener locationListener = new LocationListener() {
                    @Override public void onLocationChanged(Location location) {
                        subscriber.onNext(location);
                        Timber.i("New GPS Location - Accuracy=%d, Provider=%s, Speed=%d",
                                (int) (location.getAccuracy() + 0.5f), location.getProvider(), (int) (location.getSpeed() + 0.5f));
                    }
                    @Override public void onStatusChanged(String provider, int status, Bundle extras) {}
                    @Override public void onProviderEnabled(String provider) {}
                    @Override public void onProviderDisabled(String provider) {}
                };

                LOCATION_MANAGER.requestLocationUpdates(LOCATION_MIN_TIME_BETWEEN_UPDATES, LOCATION_MIN_DISTANCE_BETWEEN_UPDATES,
                        locationCriteria, locationListener, Looper.getMainLooper());

                // On Unsubscribe
                subscriber.add(Subscriptions.create(new Action0() {
                    @Override public void call() {
                        Timber.d("Stopping Location Services.");
                        LOCATION_MANAGER.removeUpdates(locationListener);
                    }
                }));
            }
        }).publish()
                .refCount();
    }

    /**
     * @return Observable that observes on UI Thread.
     */
    public static Observable<Location> getLocationObservable() {
        return mLocationObservable.observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * @return Observable that observes on UI Thread.
     */
    public static Observable<Speed> createSpeedObservable() {
        return mLocationObservable.map(new Func1<Location, Speed>() {
            @Override public Speed call(Location location) {
                return new Speed(location.getSpeed(), location.getAccuracy());
            }
        }).observeOn(AndroidSchedulers.mainThread());
    }
}