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.

Author: Joe Hansche

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) {