Fix #21: connect and disconnect on the background executor

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

@@ -88,19 +88,21 @@ public void unsubscribe(final ParseQuery<T> query, final SubscriptionHandling su
 
     @Override
     public void reconnect() {
-        if (webSocketClient != null) {
-            webSocketClient.close();
-        }
-        this.webSocketClient = webSocketClientFactory.createInstance(webSocketClientCallback, uri);
-        this.webSocketClient.open();
+        disconnectAsync().continueWith(new Continuation<Void, Void>() {
+            @Override
+            public Void then(Task<Void> task) throws Exception {
+                webSocketClient = webSocketClientFactory.createInstance(webSocketClientCallback, uri);
+                webSocketClient.open();
+                return null;
+            }
+        });
         userInitiatedDisconnect = false;
     }
 
     @Override
     public void disconnect() {
         if (webSocketClient != null) {
-            webSocketClient.close();
-            webSocketClient = null;
+            disconnectAsync();
             userInitiatedDisconnect = true;
         }
     }
@@ -131,6 +133,20 @@ public Void call() throws Exception {
         }, taskExecutor);
     }
 
+    private Task<Void> disconnectAsync() {
+        return Task.call(new Callable<Void>() {
+            @Override
+            public Void call() throws Exception {
+                if (webSocketClient != null) {
+                    webSocketClient.close();
+                    webSocketClient = null;
+                }
+
+                return null;
+            }
+        }, taskExecutor);
+    }
+
     private void parseMessage(String message) throws LiveQueryException {
         try {
             JSONObject jsonObject = new JSONObject(message);
@@ -227,7 +243,7 @@ private void sendSubscription(final Subscription<T> subscription) {
             public Void then(Task<String> task) throws Exception {
                 String sessionToken = task.getResult();
                 SubscribeClientOperation<T> op = new SubscribeClientOperation<>(subscription.getRequestId(), subscription.getQueryState(), sessionToken);
-                
+
                 // dispatch errors
                 sendOperationAsync(op).continueWith(new Continuation<Void, Void>() {
                     public Void then(Task<Void> task) {

ParseLiveQuery/src/test/java/com/parse/TestParseLiveQueryClient.java

@@ -13,6 +13,8 @@
 import org.robolectric.annotation.Config;
 
 import java.net.URI;
+import java.util.LinkedList;
+import java.util.Queue;
 import java.util.concurrent.Executor;
 
 import bolts.Task;
@@ -23,8 +25,8 @@
 import static org.mockito.AdditionalMatchers.and;
 import static org.mockito.AdditionalMatchers.not;
 import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.anyBoolean;
+import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.contains;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.mock;
@@ -37,6 +39,7 @@
 @Config(constants = BuildConfig.class, sdk = 21)
 public class TestParseLiveQueryClient {
 
+    private PauseableExecutor executor;
     private WebSocketClient webSocketClient;
     private WebSocketClient.WebSocketClientCallback webSocketClientCallback;
     private ParseLiveQueryClient<ParseObject> parseLiveQueryClient;
@@ -64,19 +67,16 @@ public Task<String> answer(InvocationOnMock invocation) throws Throwable {
         });
         ParseCorePlugins.getInstance().registerCurrentUserController(currentUserController);
 
+        executor = new PauseableExecutor();
+
         parseLiveQueryClient = ParseLiveQueryClient.Factory.getClient(new URI(""), new WebSocketClientFactory() {
             @Override
             public WebSocketClient createInstance(WebSocketClient.WebSocketClientCallback webSocketClientCallback, URI hostUrl) {
                 TestParseLiveQueryClient.this.webSocketClientCallback = webSocketClientCallback;
                 webSocketClient = mock(WebSocketClient.class);
                 return webSocketClient;
             }
-        }, new Executor() {
-            @Override
-            public void execute(Runnable command) {
-                command.run();
-            }
-        });
+        }, executor);
         reconnect();
     }
 
@@ -345,6 +345,16 @@ public void testEmptySessionTokenOnSubscribe() {
                 contains("\"sessionToken\":\"the token\"")));
     }
 
+    @Test
+    public void testDisconnectOnBackgroundThread() throws Exception {
+        executor.pause();
+
+        parseLiveQueryClient.disconnect();
+        verify(webSocketClient, never()).close();
+        assertTrue(executor.advanceOne());
+        verify(webSocketClient, times(1)).close();
+    }
+
     private SubscriptionHandling<ParseObject> createSubscription(ParseQuery<ParseObject> parseQuery,
             SubscriptionHandling.HandleSubscribeCallback<ParseObject> subscribeMockCallback) throws Exception {
         SubscriptionHandling<ParseObject> subscriptionHandling = parseLiveQueryClient.subscribe(parseQuery).handleSubscribe(subscribeMockCallback);
@@ -443,4 +453,39 @@ private static JSONObject createObjectDeleteMessage(int requestId, ParseObject p
         jsonObject.put("object", PointerEncoder.get().encodeRelatedObject(parseObject));
         return jsonObject;
     }
+
+    private static class PauseableExecutor implements Executor {
+        private boolean isPaused = false;
+        private final Queue<Runnable> queue = new LinkedList<>();
+
+        void pause() {
+            isPaused = true;
+        }
+
+        void unpause() {
+            if (isPaused) {
+                isPaused = false;
+
+                //noinspection StatementWithEmptyBody
+                while (advanceOne()) {
+                    // keep going
+                }
+            }
+        }
+
+        boolean advanceOne() {
+            Runnable next = queue.poll();
+            if (next != null) next.run();
+            return next != null;
+        }
+
+        @Override
+        public void execute(Runnable runnable) {
+            if (isPaused) {
+                queue.add(runnable);
+            } else {
+                runnable.run();
+            }
+        }
+    }
 }