android – 基于Observable的API和取消订阅问题

我正在尝试使用Rx- Java在 Android上创建一个位置跟踪类.我还不知道如何正确处理我的Observable的生命周期.我想要的是一个Observable,它在第一次订阅发生时开始跟踪位置,并在最后一次订阅被丢弃时停止位置跟踪.到目前为止我取得的成就是：

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);
    }
}

如您所见,我在第一个客户端订阅时使用Observable #defer初始化位置回调.我不知道这是不是一个好方法,但这是我现在提出的最好的方法.我仍然缺少的是当我的班级的最后一个客户端从我的observable取消订阅时如何停止位置更新.或者也许它在Rx中是非惯用的,因为它不明显？

我相信,这个用例应该是标准的,因此应该有一个标准/惯用的解决方案.很高兴知道它.

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();