Push scalability (#3080)

* Update status through increment
* adds support for incrementing nested keys
* fix issue when having spaces in keys for ordering
* Refactors PushController to use worker
* Adds tests for custom push queue config
* Makes PushController adapter independant
* Better logging of _PushStatus in VERBOSE

src/Push/PushQueue.js

@@ -0,0 +1,60 @@
+import { ParseMessageQueue }  from '../ParseMessageQueue';
+import rest                   from '../rest';
+import { isPushIncrementing } from './utils';
+
+const PUSH_CHANNEL = 'parse-server-push';
+const DEFAULT_BATCH_SIZE = 100;
+
+export class PushQueue {
+  parsePublisher: Object;
+  channel: String;
+  batchSize: Number;
+
+  // config object of the publisher, right now it only contains the redisURL,
+  // but we may extend it later.
+  constructor(config: any = {}) {
+    this.channel = config.channel || PUSH_CHANNEL;
+    this.batchSize = config.batchSize || DEFAULT_BATCH_SIZE;
+    this.parsePublisher = ParseMessageQueue.createPublisher(config);
+  }
+
+  static defaultPushChannel() {
+    return PUSH_CHANNEL;
+  }
+
+  enqueue(body, where, config, auth, pushStatus) {
+    const limit = this.batchSize;
+    // Order by badge (because the payload is badge dependant)
+    // and createdAt to fix the order
+    const order = isPushIncrementing(body) ? 'badge,createdAt' : 'createdAt';
+
+    return Promise.resolve().then(() => {
+      return rest.find(config,
+                       auth,
+                       '_Installation',
+                       where,
+                       {limit: 0, count: true});
+    }).then(({results, count}) => {
+      if (!results) {
+        return Promise.reject({error: 'PushController: no results in query'})
+      }
+      pushStatus.setRunning(count);
+      let skip = 0;
+      while (skip < count) {
+        const query = { where,
+          limit,
+          skip,
+          order };
+
+        const pushWorkItem = {
+          body,
+          query,
+          pushStatus: { objectId: pushStatus.objectId },
+          applicationId: config.applicationId
+        }
+        this.parsePublisher.publish(this.channel, JSON.stringify(pushWorkItem));
+        skip += limit;
+      }
+    });
+  }
+}

src/Push/PushWorker.js

@@ -0,0 +1,95 @@
+// @flow
+import deepcopy               from 'deepcopy';
+import AdaptableController    from '../Controllers/AdaptableController';
+import { master }             from '../Auth';
+import Config                 from '../Config';
+import { PushAdapter }        from '../Adapters/Push/PushAdapter';
+import rest                   from '../rest';
+import { pushStatusHandler }  from '../StatusHandler';
+import { isPushIncrementing } from './utils';
+import { ParseMessageQueue }  from '../ParseMessageQueue';
+import { PushQueue }          from './PushQueue';
+
+const UNSUPPORTED_BADGE_KEY = "unsupported";
+
+function groupByBadge(installations) {
+  return installations.reduce((map, installation) => {
+    let badge = installation.badge + '';
+    if (installation.deviceType != "ios") {
+      badge = UNSUPPORTED_BADGE_KEY;
+    }
+    map[badge] = map[badge] || [];
+    map[badge].push(installation);
+    return map;
+  }, {});
+}
+
+export class PushWorker {
+  subscriber: ?any;
+  adapter: any;
+  channel: string;
+
+  constructor(pushAdapter: PushAdapter, subscriberConfig: any = {}) {
+    AdaptableController.validateAdapter(pushAdapter, this, PushAdapter);
+    this.adapter = pushAdapter;
+
+    this.channel = subscriberConfig.channel || PushQueue.defaultPushChannel();
+    this.subscriber = ParseMessageQueue.createSubscriber(subscriberConfig);
+    if (this.subscriber) {
+      const subscriber = this.subscriber;
+      subscriber.subscribe(this.channel);
+      subscriber.on('message', (channel, messageStr) => {
+        const workItem = JSON.parse(messageStr);
+        this.run(workItem);
+      });
+    }
+  }
+
+  unsubscribe(): void {
+    if (this.subscriber) {
+      this.subscriber.unsubscribe(this.channel);
+    }
+  }
+
+  run({ body, query, pushStatus, applicationId }: any): Promise<*> {
+    const config = new Config(applicationId);
+    const auth = master(config);
+    const where  = query.where;
+    delete query.where;
+    return rest.find(config, auth, '_Installation', where, query).then(({results}) => {
+      if (results.length == 0) {
+        return;
+      }
+      return this.sendToAdapter(body, results, pushStatus, config);
+    }, err => {
+      throw err;
+    });
+  }
+
+  sendToAdapter(body: any, installations: any[], pushStatus: any, config: Config): Promise<*> {
+    pushStatus = pushStatusHandler(config, pushStatus.objectId);
+    if (!isPushIncrementing(body)) {
+      return this.adapter.send(body, installations, pushStatus.objectId).then((results) => {
+        return pushStatus.trackSent(results);
+      });
+    }
+
+    // Collect the badges to reduce the # of calls
+    const badgeInstallationsMap = groupByBadge(installations);
+
+    // Map the on the badges count and return the send result
+    const promises = Object.keys(badgeInstallationsMap).map((badge) => {
+      const payload = deepcopy(body);
+      if (badge == UNSUPPORTED_BADGE_KEY) {
+        delete payload.data.badge;
+      } else {
+        payload.data.badge = parseInt(badge);
+      }
+      const installations = badgeInstallationsMap[badge];
+      return this.sendToAdapter(payload, installations, pushStatus, config);
+    });
+    return Promise.all(promises);
+  }
+}
+
+export default PushWorker;

src/Push/utils.js

@@ -0,0 +1,30 @@
+import Parse from 'parse/node';
+
+export function isPushIncrementing(body) {
+  return body.data &&
+         body.data.badge &&
+         typeof body.data.badge == 'string' &&
+         body.data.badge.toLowerCase() == "increment"
+}
+
+/**
+ * Check whether the deviceType parameter in qury condition is valid or not.
+ * @param {Object} where A query condition
+ * @param {Array} validPushTypes An array of valid push types(string)
+ */
+export function validatePushType(where = {}, validPushTypes = []) {
+  var deviceTypeField = where.deviceType || {};
+  var deviceTypes = [];
+  if (typeof deviceTypeField === 'string') {
+    deviceTypes.push(deviceTypeField);
+  } else if (Array.isArray(deviceTypeField['$in'])) {
+    deviceTypes.concat(deviceTypeField['$in']);
+  }
+  for (var i = 0; i < deviceTypes.length; i++) {
+    var deviceType = deviceTypes[i];
+    if (validPushTypes.indexOf(deviceType) < 0) {
+      throw new Parse.Error(Parse.Error.PUSH_MISCONFIGURED,
+                            deviceType + ' is not supported push type.');
+    }
+  }
+}