Merge pull request #80 from parse-community/bugfix/issue-78

Back the subscriptions by a thread safe collection map

Author: mmimeault

ParseLiveQuery/src/main/java/com/parse/ParseLiveQueryClientImpl.java

@@ -1,8 +1,9 @@
 package com.parse;
 
 import android.util.Log;
-import android.util.SparseArray;
-
+import bolts.Continuation;
+import bolts.Task;
+import okhttp3.OkHttpClient;
 import org.json.JSONException;
 import org.json.JSONObject;
 
@@ -13,12 +14,9 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 
-import bolts.Continuation;
-import bolts.Task;
-import okhttp3.OkHttpClient;
-
 import static com.parse.Parse.checkInit;
 
 /* package */ class ParseLiveQueryClientImpl implements ParseLiveQueryClient {
@@ -28,7 +26,7 @@
     private final Executor taskExecutor;
     private final String applicationId;
     private final String clientKey;
-    private final SparseArray<Subscription<? extends ParseObject>> subscriptions = new SparseArray<>();
+    private final ConcurrentHashMap<Integer, Subscription<? extends ParseObject>> subscriptions = new ConcurrentHashMap<>();
     private final URI uri;
     private final WebSocketClientFactory webSocketClientFactory;
     private final WebSocketClient.WebSocketClientCallback webSocketClientCallback;
@@ -87,7 +85,7 @@ private static URI getDefaultUri() {
     public <T extends ParseObject> SubscriptionHandling<T> subscribe(ParseQuery<T> query) {
         int requestId = requestIdGenerator();
         Subscription<T> subscription = new Subscription<>(requestId, query);
-        subscriptions.append(requestId, subscription);
+        subscriptions.put(requestId, subscription);
 
         if (isConnected()) {
             sendSubscription(subscription);
@@ -124,8 +122,7 @@ public void connectIfNeeded() {
     @Override
     public <T extends ParseObject> void unsubscribe(final ParseQuery<T> query) {
         if (query != null) {
-            for (int i = 0; i < subscriptions.size(); i++) {
-                Subscription subscription = subscriptions.valueAt(i);
+            for (Subscription<? extends ParseObject> subscription : subscriptions.values()) {
                 if (query.equals(subscription.getQuery())) {
                     sendUnsubscription(subscription);
                 }
@@ -136,8 +133,7 @@ public <T extends ParseObject> void unsubscribe(final ParseQuery<T> query) {
     @Override
     public <T extends ParseObject> void unsubscribe(final ParseQuery<T> query, final SubscriptionHandling<T> subscriptionHandling) {
         if (query != null && subscriptionHandling != null) {
-            for (int i = 0; i < subscriptions.size(); i++) {
-                Subscription subscription = subscriptions.valueAt(i);
+            for (Subscription<? extends ParseObject> subscription : subscriptions.values()) {
                 if (query.equals(subscription.getQuery()) && subscriptionHandling.equals(subscription)) {
                     sendUnsubscription(subscription);
                 }
@@ -230,8 +226,8 @@ private void parseMessage(String message) throws LiveQueryException {
                     hasReceivedConnected = true;
                     dispatchConnected();
                     Log.v(LOG_TAG, "Connected, sending pending subscription");
-                    for (int i = 0; i < subscriptions.size(); i++) {
-                        sendSubscription(subscriptions.valueAt(i));
+                    for (Subscription<? extends ParseObject> subscription : subscriptions.values()) {
+                        sendSubscription(subscription);
                     }
                     break;
                 case "redirect":