Resolve race condition by ensuring that op=connected has been received

before sending a new subscribe event.

Fixes #46

Author: jhansche

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

@@ -38,6 +38,7 @@
     private WebSocketClient webSocketClient;
     private int requestIdCount = 1;
     private boolean userInitiatedDisconnect = false;
+    private boolean hasReceivedConnected = false;
 
     /* package */ ParseLiveQueryClientImpl() {
         this(getDefaultUri());
@@ -89,7 +90,7 @@ public <T extends ParseObject> SubscriptionHandling<T> subscribe(ParseQuery<T> q
         subscriptions.append(requestId, subscription);
 
         // TODO: differentiate between state=CONNECTED, vs received op=connected response
-        if (inAnyState(WebSocketClient.State.CONNECTED)) {
+        if (isConnected()) {
             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.");
@@ -100,6 +101,10 @@ public <T extends ParseObject> SubscriptionHandling<T> subscribe(ParseQuery<T> q
         return subscription;
     }
 
+    private boolean isConnected() {
+        return hasReceivedConnected && inAnyState(WebSocketClient.State.CONNECTED);
+    }
+
     public void connectIfNeeded() {
         switch (getWebSocketState()) {
             case CONNECTED:
@@ -151,18 +156,21 @@ public void reconnect() {
             webSocketClient.close();
         }
 
+        userInitiatedDisconnect = false;
+        hasReceivedConnected = false;
         webSocketClient = webSocketClientFactory.createInstance(webSocketClientCallback, uri);
         webSocketClient.open();
-        userInitiatedDisconnect = false;
     }
 
     @Override
     public void disconnect() {
         if (webSocketClient != null) {
-            userInitiatedDisconnect = true;
             webSocketClient.close();
             webSocketClient = null;
         }
+
+        userInitiatedDisconnect = true;
+        hasReceivedConnected = false;
     }
 
     @Override
@@ -220,6 +228,7 @@ private void parseMessage(String message) throws LiveQueryException {
 
             switch (rawOperation) {
                 case "connected":
+                    hasReceivedConnected = true;
                     dispatchConnected();
                     Log.v(LOG_TAG, "Connected, sending pending subscription");
                     for (int i = 0; i < subscriptions.size(); i++) {
@@ -371,6 +380,7 @@ private WebSocketClient.WebSocketClientCallback getWebSocketClientCallback() {
         return new WebSocketClient.WebSocketClientCallback() {
             @Override
             public void onOpen() {
+                hasReceivedConnected = false;
                 Log.v(LOG_TAG, "Socket opened");
                 ParseUser.getCurrentSessionTokenAsync().onSuccessTask(new Continuation<String, Task<Void>>() {
                     @Override
@@ -406,12 +416,14 @@ public Void then(Task<Void> task) {
             @Override
             public void onClose() {
                 Log.v(LOG_TAG, "Socket onClose");
+                hasReceivedConnected = false;
                 dispatchDisconnected();
             }
 
             @Override
             public void onError(Throwable exception) {
                 Log.e(LOG_TAG, "Socket onError", exception);
+                hasReceivedConnected = false;
                 dispatchSocketError(exception);
             }
 

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

@@ -84,7 +84,8 @@ public void tearDown() throws Exception {
     }
 
     @Test
-    public void testSubscribeBetweenSocketConnectAndConnectResponse() throws Exception {
+    public void testSubscribeAfterSocketConnectBeforeConnectedOp() throws Exception {
+        // Bug: https://github.com/parse-community/ParseLiveQuery-Android/issues/46
         ParseQuery<ParseObject> queryA = ParseQuery.getQuery("objA");
         ParseQuery<ParseObject> queryB = ParseQuery.getQuery("objB");
         clearConnection();
@@ -102,21 +103,12 @@ public void testSubscribeBetweenSocketConnectAndConnectResponse() throws Excepti
         verify(webSocketClient, times(1)).send(contains("\"op\":\"connect\""));
 
         // Now if we subscribe to queryB, we SHOULD NOT send the subscribe yet, until we get op=connected
-        SubscriptionHandling<ParseObject> subB = parseLiveQueryClient.subscribe(queryB); // TODO: fix this state
+        SubscriptionHandling<ParseObject> subB = parseLiveQueryClient.subscribe(queryB);
         verify(webSocketClient, never()).send(contains("\"op\":\"subscribe\""));
 
         // on op=connected, _then_ we should send both subscriptions
         webSocketClientCallback.onMessage(createConnectedMessage().toString());
         verify(webSocketClient, times(2)).send(contains("\"op\":\"subscribe\""));
-
-        // 1. Subscribe to queryA
-        //    - it is not connected yet, so it will trigger reconnect.
-        // 2. Socket opens & connects;  initiate op=connect
-        // 3. subscribe to queryB
-        //    - SOCKET is connected, but we haven't received op=connected yet.
-        //    - BUG: it will call sendSubscription now
-        // 4. Server responds to #2 with op=connected
-        // 5. On op=connected, we replay pending subscriptions, including the one that was already sent in #3
     }
 
     @Test