fix: LiveQuerySubscription.unsubscribe resolves promise before unsubscribing completes (#1727)

Author: dplewis

src/LiveQueryClient.js

@@ -224,19 +224,20 @@ class LiveQueryClient extends EventEmitter {
    * After calling unsubscribe you'll stop receiving events from the subscription object.
    *
    * @param {object} subscription - subscription you would like to unsubscribe from.
+   * @returns {Promise | undefined}
    */
-  unsubscribe(subscription: Object) {
+  unsubscribe(subscription: Object): ?Promise {
     if (!subscription) {
       return;
     }
-
-    this.subscriptions.delete(subscription.id);
     const unsubscribeRequest = {
       op: OP_TYPES.UNSUBSCRIBE,
       requestId: subscription.id,
     };
-    this.connectPromise.then(() => {
-      this.socket.send(JSON.stringify(unsubscribeRequest));
+    return this.connectPromise.then(() => {
+      return this.socket.send(JSON.stringify(unsubscribeRequest));
+    }).then(() => {
+      return subscription.unsubscribePromise;
     });
   }
 
@@ -400,9 +401,14 @@ class LiveQueryClient extends EventEmitter {
       }
       break;
     }
-    case OP_EVENTS.UNSUBSCRIBED:
-      // We have already deleted subscription in unsubscribe(), do nothing here
+    case OP_EVENTS.UNSUBSCRIBED: {
+      if (subscription) {
+        this.subscriptions.delete(data.requestId);
+        subscription.subscribed = false;
+        subscription.unsubscribePromise.resolve();
+      }
       break;
+    }
     default: {
       // create, update, enter, leave, delete cases
       if (!subscription) {

src/LiveQuerySubscription.js

@@ -99,6 +99,7 @@ class Subscription extends EventEmitter {
     this.query = query;
     this.sessionToken = sessionToken;
     this.subscribePromise = resolvingPromise();
+    this.unsubscribePromise = resolvingPromise();
     this.subscribed = false;
 
     // adding listener so process does not crash
@@ -115,8 +116,8 @@ class Subscription extends EventEmitter {
     return CoreManager.getLiveQueryController()
       .getDefaultLiveQueryClient()
       .then(liveQueryClient => {
-        liveQueryClient.unsubscribe(this);
         this.emit('close');
+        return liveQueryClient.unsubscribe(this);
       });
   }
 }

src/__tests__/LiveQueryClient-test.js

@@ -72,34 +72,6 @@ describe('LiveQueryClient', () => {
     liveQueryClient.open();
   });
 
-  it('can unsubscribe', async () => {
-    const liveQueryClient = new LiveQueryClient({
-      applicationId: 'applicationId',
-      serverURL: 'ws://test',
-      javascriptKey: 'javascriptKey',
-      masterKey: 'masterKey',
-      sessionToken: 'sessionToken',
-    });
-    liveQueryClient.socket = {
-      send: jest.fn(),
-    };
-    const subscription = {
-      id: 1,
-    };
-    liveQueryClient.subscriptions.set(1, subscription);
-
-    liveQueryClient.unsubscribe(subscription);
-    liveQueryClient.connectPromise.resolve();
-    expect(liveQueryClient.subscriptions.size).toBe(0);
-    await liveQueryClient.connectPromise;
-    const messageStr = liveQueryClient.socket.send.mock.calls[0][0];
-    const message = JSON.parse(messageStr);
-    expect(message).toEqual({
-      op: 'unsubscribe',
-      requestId: 1,
-    });
-  });
-
   it('can handle open / close states', () => {
     const liveQueryClient = new LiveQueryClient({
       applicationId: 'applicationId',
@@ -284,6 +256,7 @@ describe('LiveQueryClient', () => {
     });
     const subscription = new events.EventEmitter();
     subscription.subscribePromise = resolvingPromise();
+    subscription.unsubscribePromise = resolvingPromise();
 
     liveQueryClient.subscriptions.set(1, subscription);
     const data = {
@@ -295,7 +268,7 @@ describe('LiveQueryClient', () => {
       data: JSON.stringify(data),
     };
     liveQueryClient._handleWebSocketMessage(event);
-    expect(liveQueryClient.subscriptions.size).toBe(1);
+    expect(liveQueryClient.subscriptions.size).toBe(0);
   });
 
   it('can handle WebSocket error response message', async () => {
@@ -871,19 +844,28 @@ describe('LiveQueryClient', () => {
     };
     const subscription = {
       id: 1,
+      unsubscribePromise: resolvingPromise(),
     };
     liveQueryClient.subscriptions.set(1, subscription);
 
     liveQueryClient.unsubscribe(subscription);
     liveQueryClient.connectPromise.resolve();
-    expect(liveQueryClient.subscriptions.size).toBe(0);
+    expect(liveQueryClient.subscriptions.size).toBe(1);
     await liveQueryClient.connectPromise;
     const messageStr = liveQueryClient.socket.send.mock.calls[0][0];
     const message = JSON.parse(messageStr);
     expect(message).toEqual({
       op: 'unsubscribe',
       requestId: 1,
     });
+    const event = {
+      data: JSON.stringify({
+        op: 'unsubscribed',
+        requestId: 1,
+      }),
+    };
+    liveQueryClient._handleWebSocketMessage(event);
+    expect(liveQueryClient.subscriptions.size).toBe(0);
   });
 
   it('can unsubscribe without subscription', async () => {

src/__tests__/ParseLiveQuery-test.js

@@ -227,7 +227,7 @@ describe('ParseLiveQuery', () => {
     });
   });
 
-  it('should not throw on usubscribe', done => {
+  it('should not throw on usubscribe', () => {
     CoreManager.set('UserController', {
       currentUserAsync() {
         return Promise.resolve({
@@ -240,7 +240,7 @@ describe('ParseLiveQuery', () => {
     const query = new ParseQuery('ObjectType');
     query.equalTo('test', 'value');
     const subscription = new LiveQuerySubscription('0', query, 'token');
-    subscription.unsubscribe().then(done).catch(done.fail);
+    subscription.unsubscribe();
   });
 
   it('can handle LiveQuery open event', async () => {