Unit test for race condition in subscribe/connected

Author: jhansche

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

@@ -88,6 +88,7 @@ public <T extends ParseObject> SubscriptionHandling<T> subscribe(ParseQuery<T> q
         Subscription<T> subscription = new Subscription<>(requestId, query);
         subscriptions.append(requestId, subscription);
 
+        // TODO: differentiate between state=CONNECTED, vs received op=connected response
         if (inAnyState(WebSocketClient.State.CONNECTED)) {
             sendSubscription(subscription);
         } else if (userInitiatedDisconnect) {

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

@@ -1,5 +1,6 @@
 package com.parse;
 
+import org.assertj.core.api.Assertions;
 import org.json.JSONException;
 import org.json.JSONObject;
 import org.junit.After;
@@ -82,6 +83,42 @@ public void tearDown() throws Exception {
         ParsePlugins.reset();
     }
 
+    @Test
+    public void testSubscribeBetweenSocketConnectAndConnectResponse() throws Exception {
+        ParseQuery<ParseObject> queryA = ParseQuery.getQuery("objA");
+        ParseQuery<ParseObject> queryB = ParseQuery.getQuery("objB");
+        clearConnection();
+
+        // This will trigger connectIfNeeded(), which calls reconnect()
+        SubscriptionHandling<ParseObject> subA = parseLiveQueryClient.subscribe(queryA);
+
+        verify(webSocketClient, times(1)).open();
+        verify(webSocketClient, never()).send(anyString());
+
+        // Now the socket is open
+        webSocketClientCallback.onOpen();
+        when(webSocketClient.getState()).thenReturn(WebSocketClient.State.CONNECTED);
+        // and we send op=connect
+        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
+        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
     public void testSubscribeWhenSubscribedToCallback() throws Exception {
         SubscriptionHandling.HandleSubscribeCallback<ParseObject> subscribeMockCallback = mock(SubscriptionHandling.HandleSubscribeCallback.class);
@@ -459,6 +496,11 @@ private void validateSameObject(SubscriptionHandling.HandleEventCallback<ParseOb
         assertEquals(originalParseObject.getObjectId(), newParseObject.getObjectId());
     }
 
+    private void clearConnection() {
+        webSocketClient = null;
+        webSocketClientCallback = null;
+    }
+
     private void reconnect() {
         parseLiveQueryClient.reconnect();
         webSocketClientCallback.onOpen();