Fix #21: connect and disconnect on the background executor (#22)

* Fix #21: connect and disconnect on the background executor

TubeSock already uses a background thread for `open()`, but it keeps the
`close()` call on the calling thread. So `open()` is safe on the main thread,
but `close()` is not.

* Add test for #21

* Update the tests to always use a PauseableExecutor

The end result will be the same for all the other tests, since it will only
queue up tasks while it is paused, and it is unpaused by default.

To control when the tasks run, the test should first pause the executor
before the tasks get enqueued.

Author: Joe

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;
         }
     }
@@ -134,6 +136,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);
@@ -230,7 +246,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();
+            }
+        }
+    }
 }