Skip to content

Commit 5ee79d9

Browse files
author
Joe Hansche
committed
Add callbacks to be notified of connect/disconnect/error events
This enables a partial fix/workaround for #6, which otherwise wouldn't even be possible in its current state. Future changes will allow the client to automatically reconnect when the connection has been broken for any reason other than explicit disconnect().
1 parent d4f154b commit 5ee79d9

File tree

4 files changed

+128
-3
lines changed

4 files changed

+128
-3
lines changed

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,10 @@ public interface ParseLiveQueryClient {
1414

1515
void disconnect();
1616

17+
void registerListener(ParseLiveQueryClientCallbacks listener);
18+
19+
void unregisterListener(ParseLiveQueryClientCallbacks listener);
20+
1721
class Factory {
1822

1923
public static ParseLiveQueryClient getClient() {
@@ -38,5 +42,4 @@ static ParseLiveQueryClient getClient(URI uri, WebSocketClientFactory webSocketC
3842
}
3943

4044
}
41-
4245
}

‎ParseLiveQuery/src/main/java/com/parse/ParseLiveQueryClientCallbacks.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
package com.parse;
2+
3+
public interface ParseLiveQueryClientCallbacks {
4+
void onLiveQueryClientConnected(ParseLiveQueryClient client);
5+
6+
void onLiveQueryClientDisconnected(ParseLiveQueryClient client);
7+
8+
void onLiveQueryError(ParseLiveQueryClient client, LiveQueryException reason);
9+
10+
void onSocketError(ParseLiveQueryClient client, Throwable reason);
11+
}

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

Lines changed: 47 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@
99
import java.net.URI;
1010
import java.net.URISyntaxException;
1111
import java.net.URL;
12+
import java.util.ArrayList;
13+
import java.util.List;
1214
import java.util.concurrent.Callable;
1315
import java.util.concurrent.Executor;
1416

@@ -30,6 +32,8 @@
3032
private final WebSocketClientFactory webSocketClientFactory;
3133
private final WebSocketClient.WebSocketClientCallback webSocketClientCallback;
3234

35+
private final List<ParseLiveQueryClientCallbacks> mCallbacks = new ArrayList<>();
36+
3337
private WebSocketClient webSocketClient;
3438
private int requestIdCount = 1;
3539
private boolean userInitiatedDisconnect = false;
@@ -139,6 +143,16 @@ public void disconnect() {
139143
}
140144
}
141145

146+
@Override
147+
public void registerListener(ParseLiveQueryClientCallbacks listener) {
148+
mCallbacks.add(listener);
149+
}
150+
151+
@Override
152+
public void unregisterListener(ParseLiveQueryClientCallbacks listener) {
153+
mCallbacks.remove(listener);
154+
}
155+
142156
// Private methods
143157

144158
private synchronized int requestIdGenerator() {
@@ -189,6 +203,7 @@ private void parseMessage(String message) throws LiveQueryException {
189203

190204
switch (rawOperation) {
191205
case "connected":
206+
dispatchConnected();
192207
Log.v(LOG_TAG, "Connected, sending pending subscription");
193208
for (int i = 0; i < subscriptions.size(); i++) {
194209
sendSubscription(subscriptions.valueAt(i));
@@ -231,6 +246,31 @@ private void parseMessage(String message) throws LiveQueryException {
231246
}
232247
}
233248

249+
private void dispatchConnected() {
250+
for (ParseLiveQueryClientCallbacks callback : mCallbacks) {
251+
callback.onLiveQueryClientConnected(this);
252+
}
253+
}
254+
255+
private void dispatchDisconnected() {
256+
for (ParseLiveQueryClientCallbacks callback : mCallbacks) {
257+
callback.onLiveQueryClientDisconnected(this);
258+
}
259+
}
260+
261+
262+
private void dispatchServerError(LiveQueryException exc) {
263+
for (ParseLiveQueryClientCallbacks callback : mCallbacks) {
264+
callback.onLiveQueryError(this, exc);
265+
}
266+
}
267+
268+
private void dispatchSocketError(Throwable reason) {
269+
for (ParseLiveQueryClientCallbacks callback : mCallbacks) {
270+
callback.onSocketError(this, reason);
271+
}
272+
}
273+
234274
private <T extends ParseObject> void handleSubscribedEvent(JSONObject jsonObject) throws JSONException {
235275
final int requestId = jsonObject.getInt("requestId");
236276
final Subscription<T> subscription = subscriptionForRequestId(requestId);
@@ -263,9 +303,13 @@ private <T extends ParseObject> void handleErrorEvent(JSONObject jsonObject) thr
263303
String error = jsonObject.getString("error");
264304
Boolean reconnect = jsonObject.getBoolean("reconnect");
265305
final Subscription<T> subscription = subscriptionForRequestId(requestId);
306+
LiveQueryException exc = new LiveQueryException.ServerReportedException(code, error, reconnect);
307+
266308
if (subscription != null) {
267-
subscription.didEncounter(new LiveQueryException.ServerReportedException(code, error, reconnect), subscription.getQuery());
309+
subscription.didEncounter(exc, subscription.getQuery());
268310
}
311+
312+
dispatchServerError(exc);
269313
}
270314

271315
private <T extends ParseObject> Subscription<T> subscriptionForRequestId(int requestId) {
@@ -341,11 +385,13 @@ public Void then(Task<Void> task) {
341385
@Override
342386
public void onClose() {
343387
Log.v(LOG_TAG, "Socket onClose");
388+
dispatchDisconnected();
344389
}
345390

346391
@Override
347392
public void onError(Throwable exception) {
348393
Log.e(LOG_TAG, "Socket onError", exception);
394+
dispatchSocketError(exception);
349395
}
350396

351397
@Override

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

Lines changed: 66 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,12 +7,13 @@
77
import org.junit.Test;
88
import org.junit.runner.RunWith;
99
import org.mockito.ArgumentCaptor;
10-
import org.mockito.Mockito;
1110
import org.mockito.invocation.InvocationOnMock;
1211
import org.mockito.stubbing.Answer;
1312
import org.robolectric.RobolectricTestRunner;
1413
import org.robolectric.annotation.Config;
14+
import org.robolectric.util.Transcript;
1515

16+
import java.io.IOException;
1617
import java.net.URI;
1718
import java.util.LinkedList;
1819
import java.util.Queue;
@@ -398,6 +399,46 @@ public void testDisconnectOnBackgroundThread() throws Exception {
398399
verify(webSocketClient, times(1)).close();
399400
}
400401

402+
@Test
403+
public void testCallbackNotifiedOnDisconnect() throws Exception {
404+
LoggingCallbacks callbacks = new LoggingCallbacks();
405+
parseLiveQueryClient.registerListener(callbacks);
406+
callbacks.transcript.assertNoEventsSoFar();
407+
408+
webSocketClientCallback.onClose();
409+
callbacks.transcript.assertEventsSoFar("onLiveQueryClientDisconnected");
410+
}
411+
412+
@Test
413+
public void testCallbackNotifiedOnConnect() throws Exception {
414+
LoggingCallbacks callbacks = new LoggingCallbacks();
415+
parseLiveQueryClient.registerListener(callbacks);
416+
callbacks.transcript.assertNoEventsSoFar();
417+
418+
reconnect();
419+
callbacks.transcript.assertEventsSoFar("onLiveQueryClientConnected");
420+
}
421+
422+
@Test
423+
public void testCallbackNotifiedOnSocketError() throws Exception {
424+
LoggingCallbacks callbacks = new LoggingCallbacks();
425+
parseLiveQueryClient.registerListener(callbacks);
426+
callbacks.transcript.assertNoEventsSoFar();
427+
428+
webSocketClientCallback.onError(new IOException("bad things happened"));
429+
callbacks.transcript.assertEventsSoFar("onSocketError: java.io.IOException: bad things happened");
430+
}
431+
432+
@Test
433+
public void testCallbackNotifiedOnServerError() throws Exception {
434+
LoggingCallbacks callbacks = new LoggingCallbacks();
435+
parseLiveQueryClient.registerListener(callbacks);
436+
callbacks.transcript.assertNoEventsSoFar();
437+
438+
webSocketClientCallback.onMessage(createErrorMessage(1).toString());
439+
callbacks.transcript.assertEventsSoFar("onLiveQueryError: com.parse.LiveQueryException$ServerReportedException: Server reported error; code: 1, error: testError, reconnect: true");
440+
}
441+
401442
private SubscriptionHandling<ParseObject> createSubscription(ParseQuery<ParseObject> parseQuery,
402443
SubscriptionHandling.HandleSubscribeCallback<ParseObject> subscribeMockCallback) throws Exception {
403444
SubscriptionHandling<ParseObject> subscriptionHandling = parseLiveQueryClient.subscribe(parseQuery).handleSubscribe(subscribeMockCallback);
@@ -498,6 +539,30 @@ private static JSONObject createObjectDeleteMessage(int requestId, ParseObject p
498539
return jsonObject;
499540
}
500541

542+
private static class LoggingCallbacks implements ParseLiveQueryClientCallbacks {
543+
final Transcript transcript = new Transcript();
544+
545+
@Override
546+
public void onLiveQueryClientConnected(ParseLiveQueryClient client) {
547+
transcript.add("onLiveQueryClientConnected");
548+
}
549+
550+
@Override
551+
public void onLiveQueryClientDisconnected(ParseLiveQueryClient client) {
552+
transcript.add("onLiveQueryClientDisconnected");
553+
}
554+
555+
@Override
556+
public void onLiveQueryError(ParseLiveQueryClient client, LiveQueryException reason) {
557+
transcript.add("onLiveQueryError: " + reason);
558+
}
559+
560+
@Override
561+
public void onSocketError(ParseLiveQueryClient client, Throwable reason) {
562+
transcript.add("onSocketError: " + reason);
563+
}
564+
}
565+
501566
private static class PauseableExecutor implements Executor {
502567
private boolean isPaused = false;
503568
private final Queue<Runnable> queue = new LinkedList<>();

0 commit comments

Comments
 (0)