Fixes LiveQuery behavior on multiple subscriptions (#758)

* Fixes LiveQuery behavior on multiple subscriptions

* initial test and cleanup

* Massive cleanup

* improve coverage

* quick nits

Author: Amphaal

integration/server.js

@@ -9,7 +9,11 @@ const api = new ParseServer({
   appId: 'integration',
   masterKey: 'notsosecret',
   serverURL: 'http://localhost:1337/parse', // Don't forget to change to https if needed
-  cloud: `${__dirname}/cloud/main.js`
+  cloud: `${__dirname}/cloud/main.js`,
+  liveQuery: {
+    classNames: ['TestObject', 'DiffObject'],
+  },
+  startLiveQueryServer: true,
 });
 
 // Serve the Parse API on the /parse URL prefix

integration/test/ParseLiveQueryTest.js

@@ -0,0 +1,148 @@
+'use strict';
+
+const assert = require('assert');
+const clear = require('./clear');
+const Parse = require('../../node');
+
+const TestObject = Parse.Object.extend('TestObject');
+const DiffObject = Parse.Object.extend('DiffObject');
+
+describe('Parse LiveQuery', () => {
+  beforeEach((done) => {
+    Parse.initialize('integration');
+    Parse.CoreManager.set('SERVER_URL', 'http://localhost:1337/parse');
+    Parse.Storage._clear();
+    clear().then(done).catch(done.fail);
+  });
+
+  it('can subscribe to query', async () => {
+    const object = new TestObject();
+    await object.save();
+
+    const query = new Parse.Query(TestObject);
+    query.equalTo('objectId', object.id);
+    const subscription = await query.subscribe();
+
+    subscription.on('update', object => {
+      assert.equal(object.get('foo'), 'bar');
+    })
+    object.set({ foo: 'bar' });
+    await object.save();
+  });
+
+  it('can subscribe to multiple queries', async (done) => {
+    const objectA = new TestObject();
+    const objectB = new TestObject();
+    await Parse.Object.saveAll([objectA, objectB]);
+
+    const queryA = new Parse.Query(TestObject);
+    const queryB = new Parse.Query(TestObject);
+    queryA.equalTo('objectId', objectA.id);
+    queryB.equalTo('objectId', objectB.id);
+    const subscriptionA = await queryA.subscribe();
+    const subscriptionB = await queryB.subscribe();
+    let count = 0;
+    subscriptionA.on('update', object => {
+      count++;
+      assert.equal(object.get('foo'), 'bar');
+    })
+    subscriptionB.on('update', object => {
+      count++;
+      assert.equal(object.get('foo'), 'baz');
+    })
+    await objectA.save({ foo: 'bar' });
+    await objectB.save({ foo: 'baz' });
+
+    setTimeout(() => {
+      assert.equal(count, 2);
+      done();
+    }, 100);
+  });
+
+  it('can subscribe to multiple queries different class', async (done) => {
+    const objectA = new TestObject();
+    const objectB = new DiffObject();
+    await Parse.Object.saveAll([objectA, objectB]);
+
+    const queryA = new Parse.Query(TestObject);
+    const queryB = new Parse.Query(DiffObject);
+    queryA.equalTo('objectId', objectA.id);
+    queryB.equalTo('objectId', objectB.id);
+    const subscriptionA = await queryA.subscribe();
+    const subscriptionB = await queryB.subscribe();
+    let count = 0;
+    subscriptionA.on('update', object => {
+      count++;
+      assert.equal(object.get('foo'), 'bar');
+    })
+    subscriptionB.on('update', object => {
+      count++;
+      assert.equal(object.get('foo'), 'baz');
+    })
+    await objectA.save({ foo: 'bar' });
+    await objectB.save({ foo: 'baz' });
+
+    setTimeout(() => {
+      expect(count).toBe(2);
+      done();
+    }, 1000);
+  });
+
+  it('can unsubscribe to multiple queries different class', async (done) => {
+    const objectA = new TestObject();
+    const objectB = new DiffObject();
+    await Parse.Object.saveAll([objectA, objectB]);
+
+    const queryA = new Parse.Query(TestObject);
+    const queryB = new Parse.Query(DiffObject);
+    queryA.equalTo('objectId', objectA.id);
+    queryB.equalTo('objectId', objectB.id);
+    const subscriptionA = await queryA.subscribe();
+    const subscriptionB = await queryB.subscribe();
+    let count = 0;
+    subscriptionA.on('update', () => {
+      count++;
+    })
+    subscriptionB.on('update', object => {
+      count++;
+      assert.equal(object.get('foo'), 'baz');
+    })
+    subscriptionA.unsubscribe();
+    await objectA.save({ foo: 'bar' });
+    await objectB.save({ foo: 'baz' });
+
+    setTimeout(() => {
+      assert.equal(count, 1);
+      done();
+    }, 1000);
+  });
+
+  it('can unsubscribe with await to multiple queries different class', async (done) => {
+    const objectA = new TestObject();
+    const objectB = new DiffObject();
+    await Parse.Object.saveAll([objectA, objectB]);
+
+    const queryA = new Parse.Query(TestObject);
+    const queryB = new Parse.Query(DiffObject);
+    queryA.equalTo('objectId', objectA.id);
+    queryB.equalTo('objectId', objectB.id);
+    const subscriptionA = await queryA.subscribe();
+    const subscriptionB = await queryB.subscribe();
+    let count = 0;
+    subscriptionA.on('update', () => {
+      count++;
+    })
+    subscriptionB.on('update', object => {
+      count++;
+      assert.equal(object.get('foo'), 'baz');
+    })
+    await subscriptionA.unsubscribe();
+    await objectA.save({ foo: 'bar' });
+    await objectB.save({ foo: 'baz' });
+
+    setTimeout(() => {
+      assert.equal(count, 1);
+      done();
+    }, 1000);
+  });
+});

integration/test/helper.js

@@ -1,8 +1,13 @@
+const ParseServer = require('parse-server').ParseServer;
+
 jasmine.DEFAULT_TIMEOUT_INTERVAL = 5000;
 beforeAll((done) => {
   const { app } = require('../server');
-  app.listen(1337, () => {
+  const httpServer = require('http').createServer(app);
+
+  httpServer.listen(1337, () => {
     console.log('parse-server running on port 1337.');
     done();
   });
+  ParseServer.createLiveQueryServer(httpServer);
 });

src/CoreManager.js

@@ -393,10 +393,9 @@ module.exports = {
 
   setLiveQueryController(controller: any) {
     requireMethods('LiveQueryController', [
-      'subscribe',
-      'unsubscribe',
-      'open',
-      'close',
+      'setDefaultLiveQueryClient',
+      'getDefaultLiveQueryClient',
+      '_clearCachedDefaultClient',
     ], controller);
     config['LiveQueryController'] = controller;
   },

src/LiveQueryClient.js

@@ -209,10 +209,6 @@ class LiveQueryClient extends EventEmitter {
       this.socket.send(JSON.stringify(subscribeRequest));
     });
 
-    // adding listener so process does not crash
-    // best practice is for developer to register their own listener
-    subscription.on('error', () => {});
-
     return subscription;
   }
 

src/LiveQuerySubscription.js

@@ -101,12 +101,16 @@ class Subscription extends EventEmitter {
     this.id = id;
     this.query = query;
     this.sessionToken = sessionToken;
+
+    // adding listener so process does not crash
+    // best practice is for developer to register their own listener
+    this.on('error', () => {});
   }
 
   /**
-   * closes the subscription
+   * Close the subscription
    */
-  unsubscribe() {
+  unsubscribe(): Promise {
     return CoreManager.getLiveQueryController().getDefaultLiveQueryClient().then((liveQueryClient) => {
       liveQueryClient.unsubscribe(this);
       this.emit('close');