Observable-based API 및 구독 취소 문제

Question

Android에서 위치 추적을 위해 Rx-Java를 사용하려고합니다. 내가 아직 파악할 수없는 것은 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를 사용합니다. 나는 그것이 좋은 접근 방법인지는 모르겠지만, 그 순간에 내가 생각해 낸 최선의 방법이다. 제가 아직 실종 상태 인 것은, 마지막 클래스의 클라이언트가 내 관찰 가능 항목을 구독 취소했을 때 위치 업데이트를 중단하는 방법입니다. 아니면 Rx에서 낯설지 않은 무언가 일 수도 있습니다. 명백하지 않기 때문입니까?

이 사용 사례는 다소 표준적인 것이어야하므로 표준/관용적 솔루션이 있어야한다고 생각합니다. 그것을 아는 것이 기쁠 것입니다.

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