@Test public void BaseObservable_ApiClient_Connected_Dispose() { final Object object = new Object(); RxLocationFlowableOnSubscribe<Object> observable = spy(new RxLocationFlowableOnSubscribe<Object>(ctx, new Api[] {}, null) { @Override protected void onGoogleApiClientReady(GoogleApiClient apiClient, FlowableEmitter<? super Object> subscriber) { subscriber.onNext(object); } }); doAnswer(invocation -> { RxLocationBaseOnSubscribe.ApiClientConnectionCallbacks callbacks = invocation.getArgument(0); callbacks.setClient(apiClient); callbacks.onConnected(null); return apiClient; }).when(observable).createApiClient(Matchers.any(RxLocationBaseOnSubscribe.ApiClientConnectionCallbacks.class)); doReturn(true).when(apiClient).isConnected(); Flowable.create(observable, BackpressureStrategy.MISSING).subscribe().dispose(); verify(observable).onUnsubscribed(apiClient); verify(apiClient).disconnect(); }
@Override public final void subscribe(FlowableEmitter<T> emitter) throws Exception { final GoogleApiClient apiClient = createApiClient(new ApiClientConnectionCallbacks(emitter)); try { apiClient.connect(); } catch (Throwable ex) { emitter.onError(ex); } emitter.setCancellable(() -> { if (apiClient.isConnected()) { onUnsubscribed(apiClient); } apiClient.disconnect(); }); }
RxLocationFlowableOnSubscribe extends RxLocationBaseOnSubscribe<T> implements FlowableOnSubscribe<T> { @Override public final void subscribe(FlowableEmitter<T> emitter) throws Exception { final GoogleApiClient apiClient = createApiClient(new ApiClientConnectionCallbacks(emitter)); try { apiClient.connect(); } catch (Throwable ex) { emitter.onError(ex); } emitter.setCancellable(() -> { if (apiClient.isConnected()) { onUnsubscribed(apiClient); } apiClient.disconnect(); }); } }
RxLocationFlowableOnSubscribe extends RxLocationBaseOnSubscribe<T> implements FlowableOnSubscribe<T> { @Override public final void subscribe(FlowableEmitter<T> emitter) throws Exception { final GoogleApiClient apiClient = createApiClient(new ApiClientConnectionCallbacks(emitter)); try { apiClient.connect(); } catch (Throwable ex) { emitter.onError(ex); } emitter.setCancellable(() -> { if (apiClient.isConnected()) { onUnsubscribed(apiClient); } apiClient.disconnect(); }); } protected  RxLocationFlowableOnSubscribe(@NonNull RxLocation rxLocation, Long timeout, TimeUnit timeUnit); protected  RxLocationFlowableOnSubscribe(@NonNull Context ctx, @NonNull Api<? extends Api.ApiOptions.NotRequiredOptions>[] services, Scope[] scopes); }
RxLocationFlowableOnSubscribe extends RxLocationBaseOnSubscribe<T> implements FlowableOnSubscribe<T> { @Override public final void subscribe(FlowableEmitter<T> emitter) throws Exception { final GoogleApiClient apiClient = createApiClient(new ApiClientConnectionCallbacks(emitter)); try { apiClient.connect(); } catch (Throwable ex) { emitter.onError(ex); } emitter.setCancellable(() -> { if (apiClient.isConnected()) { onUnsubscribed(apiClient); } apiClient.disconnect(); }); } protected  RxLocationFlowableOnSubscribe(@NonNull RxLocation rxLocation, Long timeout, TimeUnit timeUnit); protected  RxLocationFlowableOnSubscribe(@NonNull Context ctx, @NonNull Api<? extends Api.ApiOptions.NotRequiredOptions>[] services, Scope[] scopes); @Override final void subscribe(FlowableEmitter<T> emitter); }
RxLocationFlowableOnSubscribe extends RxLocationBaseOnSubscribe<T> implements FlowableOnSubscribe<T> { @Override public final void subscribe(FlowableEmitter<T> emitter) throws Exception { final GoogleApiClient apiClient = createApiClient(new ApiClientConnectionCallbacks(emitter)); try { apiClient.connect(); } catch (Throwable ex) { emitter.onError(ex); } emitter.setCancellable(() -> { if (apiClient.isConnected()) { onUnsubscribed(apiClient); } apiClient.disconnect(); }); } protected  RxLocationFlowableOnSubscribe(@NonNull RxLocation rxLocation, Long timeout, TimeUnit timeUnit); protected  RxLocationFlowableOnSubscribe(@NonNull Context ctx, @NonNull Api<? extends Api.ApiOptions.NotRequiredOptions>[] services, Scope[] scopes); @Override final void subscribe(FlowableEmitter<T> emitter);  }