Merge pull request #3 from wondrous-io/unswallow-subscribe-errors

Unswallow subscribe errors

Author: mmimeault

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

@@ -10,6 +10,23 @@ private LiveQueryException(String detailMessage) {
         super(detailMessage);
     }
 
+    private LiveQueryException(String detailMessage, Throwable cause) {
+        super(detailMessage, cause);
+    }
+
+    private LiveQueryException(Throwable cause) {
+        super(cause);
+    }
+
+    /**
+     * An error that is reported when any other unknown {@link RuntimeException} occurs unexpectedly.
+     */
+    public static class UnknownException extends LiveQueryException {
+        /* package */ UnknownException(String detailMessage, RuntimeException cause) {
+            super(detailMessage, cause);
+        }
+    }
+
     /**
      * An error that is reported when the server returns a response that cannot be parsed.
      */

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

@@ -222,11 +222,26 @@ private Subscription<T> subscriptionForRequestId(int requestId) {
     }
 
     private void sendSubscription(final Subscription<T> subscription) {
-        ParseUser.getCurrentSessionTokenAsync().onSuccessTask(new Continuation<String, Task<Void>>() {
+        ParseUser.getCurrentSessionTokenAsync().onSuccess(new Continuation<String, Void>() {
             @Override
-            public Task<Void> then(Task<String> task) throws Exception {
+            public Void then(Task<String> task) throws Exception {
                 String sessionToken = task.getResult();
-                return sendOperationAsync(new SubscribeClientOperation<>(subscription.getRequestId(), subscription.getQueryState(), sessionToken));
+                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) {
+                        Exception error = task.getError();
+                        if (error != null) {
+                            if (error instanceof RuntimeException) {
+                                subscription.didEncounter(new LiveQueryException.UnknownException(
+                                        "Error when subscribing", (RuntimeException) error), subscription.getQuery());
+                            }
+                        }
+                        return null;
+                    }
+                });
+                return null;
             }
         });
     }

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

@@ -18,14 +18,17 @@
 import bolts.Task;
 
 import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertNotNull;
 import static junit.framework.Assert.assertTrue;
 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.contains;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -110,6 +113,34 @@ public void testUnsubscribeWhenSubscribedToCallback() throws Exception {
         verify(unsubscribeMockCallback, times(1)).onUnsubscribe(parseQuery);
     }
 
+    @Test
+    public void testErrorWhileSubscribing() throws Exception {
+        ParseQuery.State state = mock(ParseQuery.State.class);
+        when(state.toJSON(any(ParseEncoder.class))).thenThrow(new RuntimeException("forced error"));
+
+        ParseQuery.State.Builder builder = mock(ParseQuery.State.Builder.class);
+        when(builder.build()).thenReturn(state);
+        ParseQuery query = mock(ParseQuery.class);
+        when(query.getBuilder()).thenReturn(builder);
+
+        SubscriptionHandling handling = parseLiveQueryClient.subscribe(query);
+
+        SubscriptionHandling.HandleErrorCallback<ParseObject> errorMockCallback = mock(SubscriptionHandling.HandleErrorCallback.class);
+        handling.handleError(errorMockCallback);
+
+        // Trigger a re-subscribe
+        webSocketClientCallback.onMessage(createConnectedMessage().toString());
+
+        // This will never get a chance to call op=subscribe, because an exception was thrown
+        verify(webSocketClient, never()).send(anyString());
+
+        ArgumentCaptor<LiveQueryException> errorCaptor = ArgumentCaptor.forClass(LiveQueryException.class);
+        verify(errorMockCallback, times(1)).onError(eq(query), errorCaptor.capture());
+
+        assertEquals("Error when subscribing", errorCaptor.getValue().getMessage());
+        assertNotNull(errorCaptor.getValue().getCause());
+    }
+
     @Test
     public void testErrorWhenSubscribedToCallback() throws Exception {
         ParseQuery<ParseObject> parseQuery = new ParseQuery<>("test");