Revert the async disconnect change (#42)

* Revert async-disconnect

This reverts commit 3b0041d

The theory is that using OkHttp now, all networking is handled on the correct
thread automatically.

Closes #37

* Clarify the logic of initiating a connection during subscribe()

This also creates a new connectIfNeeded() method, which can be used to
initiate a connection if not already connected or connecting.

The original intent was to connectIfNeeded() during subscribe. While some may
argue that's not necessary, and the client should instead be explicit about
what is intended, this change preserves that original intention.

As it stands, the only way to queue up pending subscriptions _without_
initiating a connection would be to call disconnect() first, even though the
client hasn't been connected yet. That would set the user-initiated flag,
which would therefore refuse to auto-connect during subscribe().

* Remove PausableExecutor

Also introduces the ImmediateExecutor (same as what we used to have, before
switching to PausableExecutor). Similar to, but simpler than,
`bolts.BoltsExecutors#immediate()`.

Author: Joe

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

@@ -10,6 +10,8 @@ public interface ParseLiveQueryClient {
 
     <T extends ParseObject> void unsubscribe(final ParseQuery<T> query, final SubscriptionHandling<T> subscriptionHandling);
 
+    void connectIfNeeded();
+
     void reconnect();
 
     void disconnect();

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

@@ -10,6 +10,7 @@
 import java.net.URISyntaxException;
 import java.net.URL;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
@@ -86,18 +87,39 @@ public <T extends ParseObject> SubscriptionHandling<T> subscribe(ParseQuery<T> q
         int requestId = requestIdGenerator();
         Subscription<T> subscription = new Subscription<>(requestId, query);
         subscriptions.append(requestId, subscription);
-        if (webSocketClient == null || (webSocketClient.getState() != WebSocketClient.State.CONNECTING && webSocketClient.getState() != WebSocketClient.State.CONNECTED)) {
-            if (!userInitiatedDisconnect) {
-                reconnect();
-            } else {
-                Log.w(LOG_TAG, "Warning: The client was explicitly disconnected! You must explicitly call .reconnect() in order to process your subscriptions.");
-            }
-        } else if (webSocketClient.getState() == WebSocketClient.State.CONNECTED) {
+
+        if (inAnyState(WebSocketClient.State.CONNECTED)) {
             sendSubscription(subscription);
+        } else if (userInitiatedDisconnect) {
+            Log.w(LOG_TAG, "Warning: The client was explicitly disconnected! You must explicitly call .reconnect() in order to process your subscriptions.");
+        } else {
+            connectIfNeeded();
         }
+
         return subscription;
     }
 
+    public void connectIfNeeded() {
+        switch (getWebSocketState()) {
+            case CONNECTED:
+                // nothing to do
+                break;
+            case CONNECTING:
+                // just wait for it to finish connecting
+                break;
+
+            case NONE:
+            case DISCONNECTING:
+            case DISCONNECTED:
+                reconnect();
+                break;
+
+            default:
+
+                break;
+        }
+    }
+
     @Override
     public <T extends ParseObject> void unsubscribe(final ParseQuery<T> query) {
         if (query != null) {
@@ -124,22 +146,21 @@ public <T extends ParseObject> void unsubscribe(final ParseQuery<T> query, final
 
     @Override
     public void reconnect() {
-        disconnectAsync().continueWith(new Continuation<Void, Void>() {
-            @Override
-            public Void then(Task<Void> task) throws Exception {
-                webSocketClient = webSocketClientFactory.createInstance(webSocketClientCallback, uri);
-                webSocketClient.open();
-                return null;
-            }
-        });
+        if (webSocketClient != null) {
+            webSocketClient.close();
+        }
+
+        webSocketClient = webSocketClientFactory.createInstance(webSocketClientCallback, uri);
+        webSocketClient.open();
         userInitiatedDisconnect = false;
     }
 
     @Override
     public void disconnect() {
         if (webSocketClient != null) {
             userInitiatedDisconnect = true;
-            disconnectAsync();
+            webSocketClient.close();
+            webSocketClient = null;
         }
     }
 
@@ -159,6 +180,15 @@ private synchronized int requestIdGenerator() {
         return requestIdCount++;
     }
 
+    private WebSocketClient.State getWebSocketState() {
+        WebSocketClient.State state = webSocketClient == null ? null : webSocketClient.getState();
+        return state == null ? WebSocketClient.State.NONE : state;
+    }
+
+    private boolean inAnyState(WebSocketClient.State... states) {
+        return Arrays.asList(states).contains(getWebSocketState());
+    }
+
     private Task<Void> handleOperationAsync(final String message) {
         return Task.call(new Callable<Void>() {
             public Void call() throws Exception {
@@ -182,20 +212,6 @@ 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);

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

@@ -0,0 +1,10 @@
+package com.parse;
+
+import java.util.concurrent.Executor;
+
+class ImmediateExecutor implements Executor {
+    @Override
+    public void execute(Runnable runnable) {
+        runnable.run();
+    }
+}

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

@@ -15,9 +15,6 @@
 
 import java.io.IOException;
 import java.net.URI;
-import java.util.LinkedList;
-import java.util.Queue;
-import java.util.concurrent.Executor;
 
 import bolts.Task;
 
@@ -41,7 +38,6 @@
 @Config(constants = BuildConfig.class, sdk = 21)
 public class TestParseLiveQueryClient {
 
-    private PauseableExecutor executor;
     private WebSocketClient webSocketClient;
     private WebSocketClient.WebSocketClientCallback webSocketClientCallback;
     private ParseLiveQueryClient parseLiveQueryClient;
@@ -69,16 +65,14 @@ 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;
             }
-        }, executor);
+        }, new ImmediateExecutor());
         reconnect();
     }
 
@@ -389,16 +383,6 @@ 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();
-    }
-
     @Test
     public void testCallbackNotifiedOnUnexpectedDisconnect() throws Exception {
         LoggingCallbacks callbacks = new LoggingCallbacks();
@@ -580,41 +564,6 @@ public void onSocketError(ParseLiveQueryClient client, Throwable reason) {
         }
     }
 
-    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();
-            }
-        }
-    }
-
     @ParseClassName("MockA")
     static class MockClassA extends ParseObject {
     }