Fix for multiple subscribers on Observables. Fixed bug which did not handle Exceptions after successful resolution. Added tests for resolution handling with multiple subscribers. Added Transformer to help with resuming with another RxFit Observable after an Exception, but propagate an unsuccessful resolution.

Author: patloew

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Version 1.1.1
+
+* Fix for multiple subscribers on RxFit Observables.
+* Fixed bug which did not handle Exceptions after successful resolution.
+* Added RxFit.OnExceptionResumeNext Transformer.
+
 ## Version 1.1.0
 
 * BREAKING CHANGE: Removed PermissionRequiredException in favor of SecurityException

README.md

@@ -1,6 +1,6 @@
 # Reactive Fit API Library for Android
 
-[![Build Status](https://travis-ci.org/patloew/RxFit.svg?branch=master)](https://travis-ci.org/patloew/RxFit) [![API](https://img.shields.io/badge/API-9%2B-brightgreen.svg?style=flat)](https://android-arsenal.com/api?level=9)
+[![Build Status](https://travis-ci.org/patloew/RxFit.svg?branch=master)](https://travis-ci.org/patloew/RxFit) [ ![Download](https://api.bintray.com/packages/patloew/maven/com.patloew.rxfit/images/download.svg) ](https://bintray.com/patloew/maven/com.patloew.rxfit/_latestVersion) [![API](https://img.shields.io/badge/API-9%2B-brightgreen.svg?style=flat)](https://android-arsenal.com/api?level=9)
 
 This library wraps the Fit API in [RxJava](https://github.com/ReactiveX/RxJava) Observables. No more managing GoogleApiClients! Also, the authorization process for using fitness data is handled by the lib.
 

library/build.gradle

@@ -3,7 +3,7 @@ apply plugin: 'com.jfrog.bintray'
 apply plugin: 'com.github.dcendents.android-maven'
 
 group = 'com.patloew.rxfit'
-version = '1.1.0'
+version = '1.1.1'
 project.archivesBaseName = 'rxfit'
 
 android {
@@ -13,8 +13,8 @@ android {
     defaultConfig {
         minSdkVersion 9
         targetSdkVersion 23
-        versionCode 2
-        versionName "1.1.0"
+        versionCode 3
+        versionName "1.1.1"
     }
     buildTypes {
         release {

library/src/main/java/com/patloew/rxfit/BaseObservable.java

@@ -14,8 +14,10 @@
 import com.google.android.gms.common.api.ResultCallback;
 import com.google.android.gms.common.api.Scope;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import rx.Observable;
@@ -44,17 +46,16 @@
  *
  */
 public abstract class BaseObservable<T> implements Observable.OnSubscribe<T> {
-    private static final List<BaseObservable> observableList = new ArrayList<>();
+    private static final Set<BaseObservable> observableSet = new HashSet<>();
 
     private final Context ctx;
     private final Api<? extends Api.ApiOptions.NotRequiredOptions>[] services;
     private final Scope[] scopes;
     private final boolean handleResolution;
-    private GoogleApiClient apiClient;
-    Subscriber<? super T> subscriber;
+    private final Long timeoutTime;
+    private final TimeUnit timeoutUnit;
 
-    protected final Long timeoutTime;
-    protected final TimeUnit timeoutUnit;
+    private final HashMap<GoogleApiClient, Subscriber<? super T>> subscriptionInfoHashMap = new HashMap<>();
 
     protected BaseObservable(@NonNull RxFit rxFit, Long timeout, TimeUnit timeUnit) {
         this.ctx = rxFit.getContext();
@@ -63,11 +64,11 @@ protected BaseObservable(@NonNull RxFit rxFit, Long timeout, TimeUnit timeUnit)
         handleResolution = true;
 
         if(timeout != null && timeUnit != null) {
-            this.timeoutTime = RxFit.getTimeout(timeout);
-            this.timeoutUnit = RxFit.getTimeoutUnit(timeUnit);
+            this.timeoutTime = timeout;
+            this.timeoutUnit = timeUnit;
         } else {
-            this.timeoutTime = RxFit.getTimeout(null);
-            this.timeoutUnit = RxFit.getTimeoutUnit(null);
+            this.timeoutTime = RxFit.getDefaultTimeout();
+            this.timeoutUnit = RxFit.getDefaultTimeoutUnit();
         }
     }
 
@@ -80,7 +81,7 @@ protected BaseObservable(@NonNull Context ctx, @NonNull Api<? extends Api.ApiOpt
         timeoutUnit = null;
     }
 
-    protected <T extends Result> void setupFitnessPendingResult(PendingResult<T> pendingResult, ResultCallback<? super T> resultCallback) {
+    protected final <T extends Result> void setupFitnessPendingResult(PendingResult<T> pendingResult, ResultCallback<? super T> resultCallback) {
         if(timeoutTime != null && timeoutUnit != null) {
             pendingResult.setResultCallback(resultCallback, timeoutTime, timeoutUnit);
         } else {
@@ -89,10 +90,9 @@ protected <T extends Result> void setupFitnessPendingResult(PendingResult<T> pen
     }
 
     @Override
-    public void call(Subscriber<? super T> subscriber) {
-        this.subscriber = subscriber;
-
-        apiClient = createApiClient(subscriber);
+    public final void call(Subscriber<? super T> subscriber) {
+        final GoogleApiClient apiClient = createApiClient(subscriber);
+        subscriptionInfoHashMap.put(apiClient, subscriber);
 
         try {
             apiClient.connect();
@@ -107,12 +107,14 @@ public void call() {
                     onUnsubscribed(apiClient);
                     apiClient.disconnect();
                 }
+
+                subscriptionInfoHashMap.remove(apiClient);
             }
         }));
     }
 
 
-    protected GoogleApiClient createApiClient(Subscriber<? super T> subscriber) {
+    GoogleApiClient createApiClient(Subscriber<? super T> subscriber) {
 
         ApiClientConnectionCallbacks apiClientConnectionCallbacks = new ApiClientConnectionCallbacks(subscriber);
 
@@ -173,7 +175,7 @@ public void onConnectionSuspended(int cause) {
         @Override
         public void onConnectionFailed(ConnectionResult connectionResult) {
             if(handleResolution && connectionResult.hasResolution()) {
-                observableList.add(BaseObservable.this);
+                observableSet.add(BaseObservable.this);
 
                 if(!ResolutionActivity.isResolutionShown()) {
                     Intent intent = new Intent(ctx, ResolutionActivity.class);
@@ -192,16 +194,22 @@ public void setClient(GoogleApiClient client) {
     }
 
     static void onResolutionResult(int resultCode, ConnectionResult connectionResult) {
-        for(BaseObservable observable : observableList) {
-            if(!observable.subscriber.isUnsubscribed()) {
-                if (resultCode == Activity.RESULT_OK && observable.apiClient != null) {
-                    observable.apiClient.connect();
-                } else {
-                    observable.subscriber.onError(new GoogleAPIConnectionException("Error connecting to GoogleApiClient, resolution was not successful.", connectionResult));
+        for(BaseObservable observable : observableSet) {
+            for (Map.Entry<GoogleApiClient, Subscriber> entry : (Set<Map.Entry<GoogleApiClient, Subscriber>>) observable.subscriptionInfoHashMap.entrySet()) {
+                if (!entry.getValue().isUnsubscribed()) {
+                    if (resultCode == Activity.RESULT_OK) {
+                        try {
+                            entry.getKey().connect();
+                        } catch (Throwable ex) {
+                            entry.getValue().onError(ex);
+                        }
+                    } else {
+                        entry.getValue().onError(new GoogleAPIConnectionException("Error connecting to GoogleApiClient, resolution was not successful.", connectionResult));
+                    }
                 }
             }
         }
 
-        observableList.clear();
+        observableSet.clear();
     }
 }

library/src/main/java/com/patloew/rxfit/GoogleAPIConnectionException.java

@@ -27,4 +27,12 @@ public class GoogleAPIConnectionException extends RuntimeException {
     public ConnectionResult getConnectionResult() {
         return connectionResult;
     }
+
+    public boolean wasResolutionUnsuccessful() {
+        if(connectionResult != null) {
+            return connectionResult.hasResolution();
+        } else {
+            return false;
+        }
+    }
 }

library/src/main/java/com/patloew/rxfit/RxFit.java

@@ -3,7 +3,6 @@
 import android.app.PendingIntent;
 import android.content.Context;
 import android.support.annotation.NonNull;
-import android.support.annotation.Nullable;
 import android.support.annotation.RequiresPermission;
 
 import com.google.android.gms.common.api.Api;
@@ -34,6 +33,8 @@
 
 import rx.Completable;
 import rx.Observable;
+import rx.exceptions.Exceptions;
+import rx.functions.Func1;
 
 /* Factory for Google Fit API observables. Make sure to include all the APIs
  * and Scopes that you need for your app. Also make sure to have the Location
@@ -106,14 +107,12 @@ Scope[] getScopes() {
         return scopes;
     }
 
-    @Nullable
-    static Long getTimeout(@Nullable Long timeout) {
-        return timeout != null ? timeout : timeoutTime;
+    static Long getDefaultTimeout() {
+        return timeoutTime;
     }
 
-    @Nullable
-    static TimeUnit getTimeoutUnit(@Nullable TimeUnit timeUnit) {
-        return timeUnit != null ? timeUnit : timeoutUnit;
+    static TimeUnit getDefaultTimeoutUnit() {
+        return timeoutUnit;
     }
 
 
@@ -446,4 +445,38 @@ public static Observable<List<Session>> stop(@NonNull String identifier, long ti
 
     }
 
+
+    /* Transformer that behaves like onExceptionResumeNext(Observable o), but propagates
+     * a GoogleAPIConnectionException, which was caused by an unsuccessful resolution.
+     * This can be helpful if you want to resume with another RxFit Observable when
+     * an Exception occurs, but don't want to show the resolution dialog multiple times.
+     *
+     * An example use case: Fetch fitness data with server queries enabled, but provide
+     * a timeout. When an exception occurs (e.g. timeout), switch to cached fitness data.
+     * Using this Transformer prevents showing the authorization dialog twice, if the user
+     * denys access for the first read. See MainActivity in sample project.
+     */
+    public static class OnExceptionResumeNext<T, R extends T> implements Observable.Transformer<T, T> {
+
+        private final Observable<R> other;
+
+        public OnExceptionResumeNext(Observable<R> other) {
+            this.other = other;
+        }
+
+        @Override
+        public Observable<T> call(Observable<T> source) {
+            return source.onErrorResumeNext(new Func1<Throwable, Observable<R>>() {
+                @Override
+                public Observable<R> call(Throwable throwable) {
+                    if (!(throwable instanceof Exception) || (throwable instanceof GoogleAPIConnectionException && ((GoogleAPIConnectionException) throwable).wasResolutionUnsuccessful())) {
+                        Exceptions.propagate(throwable);
+                    }
+
+                    return other;
+                }
+            });
+        }
+    }
+
 }