feat: Upgrade Redis 3 to 4 for LiveQuery (#8333)

src/Adapters/PubSub/RedisPubSub.js

@@ -2,12 +2,12 @@ import { createClient } from 'redis';
 
 function createPublisher({ redisURL, redisOptions = {} }): any {
   redisOptions.no_ready_check = true;
-  return createClient(redisURL, redisOptions);
+  return createClient({ url: redisURL, ...redisOptions });
 }
 
 function createSubscriber({ redisURL, redisOptions = {} }): any {
   redisOptions.no_ready_check = true;
-  return createClient(redisURL, redisOptions);
+  return createClient({ url: redisURL, ...redisOptions });
 }
 
 const RedisPubSub = {

src/Controllers/LiveQueryController.js

@@ -21,6 +21,10 @@ export class LiveQueryController {
     this.liveQueryPublisher = new ParseCloudCodePublisher(config);
   }
 
+  connect() {
+    return this.liveQueryPublisher.connect();
+  }
+
   onAfterSave(
     className: string,
     currentObject: any,

src/LiveQuery/ParseCloudCodePublisher.js

@@ -11,6 +11,15 @@ class ParseCloudCodePublisher {
     this.parsePublisher = ParsePubSub.createPublisher(config);
   }
 
+  async connect() {
+    if (typeof this.parsePublisher.connect === 'function') {
+      if (this.parsePublisher.isOpen) {
+        return;
+      }
+      return Promise.resolve(this.parsePublisher.connect());
+    }
+  }
+
   onCloudCodeAfterSave(request: any): void {
     this._onCloudCodeMessage(Parse.applicationId + 'afterSave', request);
   }

src/LiveQuery/ParseLiveQueryServer.js

@@ -73,15 +73,25 @@ class ParseLiveQueryServer {
       parseWebsocket => this._onConnect(parseWebsocket),
       config
     );
-
-    // Initialize subscriber
     this.subscriber = ParsePubSub.createSubscriber(config);
-    this.subscriber.subscribe(Parse.applicationId + 'afterSave');
-    this.subscriber.subscribe(Parse.applicationId + 'afterDelete');
-    this.subscriber.subscribe(Parse.applicationId + 'clearCache');
-    // Register message handler for subscriber. When publisher get messages, it will publish message
-    // to the subscribers and the handler will be called.
-    this.subscriber.on('message', (channel, messageStr) => {
+    if (!this.subscriber.connect) {
+      this.connect();
+    }
+  }
+
+  async connect() {
+    if (this.subscriber.isOpen) {
+      return;
+    }
+    if (typeof this.subscriber.connect === 'function') {
+      await Promise.resolve(this.subscriber.connect());
+    } else {
+      this.subscriber.isOpen = true;
+    }
+    this._createSubscribers();
+  }
+  _createSubscribers() {
+    const messageRecieved = (channel, messageStr) => {
       logger.verbose('Subscribe message %j', messageStr);
       let message;
       try {
@@ -102,7 +112,12 @@ class ParseLiveQueryServer {
       } else {
         logger.error('Get message %s from unknown channel %j', message, channel);
       }
-    });
+    };
+    this.subscriber.on('message', (channel, messageStr) => messageRecieved(channel, messageStr));
+    for (const field of ['afterSave', 'afterDelete', 'clearCache']) {
+      const channel = `${Parse.applicationId}${field}`;
+      this.subscriber.subscribe(channel, messageStr => messageRecieved(channel, messageStr));
+    }
   }
 
   // Message is the JSON object from publisher. Message.currentParseObject is the ParseObject JSON after changes.

src/ParseServer.js

@@ -77,7 +77,12 @@ class ParseServer {
 
     const allControllers = controllers.getControllers(options);
 
-    const { loggerController, databaseController, hooksController } = allControllers;
+    const {
+      loggerController,
+      databaseController,
+      hooksController,
+      liveQueryController,
+    } = allControllers;
     this.config = Config.put(Object.assign({}, options, allControllers));
 
     logging.setLogger(loggerController);
@@ -98,6 +103,7 @@ class ParseServer {
         ) {
           startupPromises.push(options.cacheAdapter.connect());
         }
+        startupPromises.push(liveQueryController.connect());
         await Promise.all(startupPromises);
         if (serverStartComplete) {
           serverStartComplete();
@@ -263,7 +269,7 @@ class ParseServer {
    * @param {Function} callback called when the server has started
    * @returns {ParseServer} the parse server instance
    */
-  start(options: ParseServerOptions, callback: ?() => void) {
+  async start(options: ParseServerOptions, callback: ?() => void) {
     const app = express();
     if (options.middleware) {
       let middleware;
@@ -307,7 +313,7 @@ class ParseServer {
     this.server = server;
 
     if (options.startLiveQueryServer || options.liveQueryServerOptions) {
-      this.liveQueryServer = ParseServer.createLiveQueryServer(
+      this.liveQueryServer = await ParseServer.createLiveQueryServer(
         server,
         options.liveQueryServerOptions,
         options
@@ -338,9 +344,9 @@ class ParseServer {
    * @param {Server} httpServer an optional http server to pass
    * @param {LiveQueryServerOptions} config options for the liveQueryServer
    * @param {ParseServerOptions} options options for the ParseServer
-   * @returns {ParseLiveQueryServer} the live query server instance
+   * @returns {Promise<ParseLiveQueryServer>} the live query server instance
    */
-  static createLiveQueryServer(
+  static async createLiveQueryServer(
     httpServer,
     config: LiveQueryServerOptions,
     options: ParseServerOptions
@@ -350,7 +356,9 @@ class ParseServer {
       httpServer = require('http').createServer(app);
       httpServer.listen(config.port);
     }
-    return new ParseLiveQueryServer(httpServer, config, options);
+    const server = new ParseLiveQueryServer(httpServer, config, options);
+    await server.connect();
+    return server;
   }
 
   static verifyServerUrl(callback) {