Researched API and Subscription Issue

Question

I am trying to use Rx-Java to create a location tracking class on Android. I cannot understand how to properly handle the life cycle of my Observer. I want to have an Observable that starts tracking location on the first subscription and stops tracking location when canceling the last subscription. So far I have achieved this:

public class LocationObservable implements GooglePlayServicesClient.ConnectionCallbacks, GooglePlayServicesClient.OnConnectionFailedListener, LocationListener { private LocationClient locationClient; private final PublishSubject<Location> latestLocation = PublishSubject.create(); public final Observable<Location> locationObservable = Observable.defer(() -> { if(!locationClient.isConnected() && !locationClient.isConnecting()) { locationClient.connect(); } return latestLocation.asObservable().scan((prev, curr) -> { if (Math.abs(prev.getLatitude() - curr.getLatitude()) > 0.000001 || Math.abs(prev.getLongitude() - curr.getLongitude()) > 0.000001) return curr; else return prev; }).distinctUntilChanged();}); public LocationObservable(Context context) { locationClient = new LocationClient(context, this, this); } @Override public void onConnected(Bundle bundle) { latestLocation.onNext(locationClient.getLastLocation()); } @Override public void onDisconnected() { latestLocation.onCompleted(); } @Override public void onConnectionFailed(ConnectionResult connectionResult) { latestLocation.onError(new Exception(connectionResult.toString())); } @Override public void onLocationChanged(Location location) { latestLocation.onNext(location); } } 

As you can see, I use Observable#defer to initiate calls when the first client signs up. I do not know if this is a good approach, but this is the best that I have come up with at the moment. What I still lack is how to stop location updates when the last client of my class does not subscribe from my observable. Or maybe this is something non-idiomatic in Rx, since it is not obvious?

I believe that this use case should be fairly standard, and therefore there should be a standard / idiomatic solution for it. I would be glad to know about it.

Answer 1

 private LocationClient locationClient; private final Observable<Integer> locationObservable = Observable .create(new OnSubscribe<Integer>() { @Override public void call(Subscriber<? super Integer> subscriber) { locationClient.connect(); subscriber.add(Subscriptions.create(new Action0() { @Override public void call() { if (locationClient.isConnected() || locationClient.isConnecting()) { locationClient.disconnect(); } } })); } }).multicast(PublishSubject.<Integer> create()).refCount();