Batching Oplog Entries & DDP messages

packages/ddp-client/common/livedata_connection.js

@@ -1607,9 +1607,9 @@ export class Connection {
   }
 
   onMessage(raw_msg) {
-    let msg;
+    let messages;
     try {
-      msg = DDPCommon.parseDDP(raw_msg);
+      messages = DDPCommon.parseDDP(raw_msg);
     } catch (e) {
       Meteor._debug('Exception while parsing DDP', e);
       return;
@@ -1621,45 +1621,51 @@ export class Connection {
       this._heartbeat.messageReceived();
     }
 
-    if (msg === null || !msg.msg) {
-      if(!msg || !msg.testMessageOnConnect) {
-        if (Object.keys(msg).length === 1 && msg.server_id) return;
-        Meteor._debug('discarding invalid livedata message', msg);
+    messages = Array.isArray(messages) ? messages : [messages];
+
+    for (let i = 0; i < messages.length; i++) {
+      let msg = messages[i];
+
+      if (msg === null || !msg.msg) {
+        if(!msg || !msg.testMessageOnConnect) {
+          if (Object.keys(msg).length === 1 && msg.server_id) return;
+          Meteor._debug('discarding invalid livedata message', msg);
+        }
+        return;
       }
-      return;
-    }
 
-    if (msg.msg === 'connected') {
-      this._version = this._versionSuggestion;
-      this._livedata_connected(msg);
-      this.options.onConnected();
-    } else if (msg.msg === 'failed') {
-      if (this._supportedDDPVersions.indexOf(msg.version) >= 0) {
-        this._versionSuggestion = msg.version;
-        this._stream.reconnect({ _force: true });
+      if (msg.msg === 'connected') {
+        this._version = this._versionSuggestion;
+        this._livedata_connected(msg);
+        this.options.onConnected();
+      } else if (msg.msg === 'failed') {
+        if (this._supportedDDPVersions.indexOf(msg.version) >= 0) {
+          this._versionSuggestion = msg.version;
+          this._stream.reconnect({ _force: true });
+        } else {
+          const description =
+            'DDP version negotiation failed; server requested version ' +
+            msg.version;
+          this._stream.disconnect({ _permanent: true, _error: description });
+          this.options.onDDPVersionNegotiationFailure(description);
+        }
+      } else if (msg.msg === 'ping' && this.options.respondToPings) {
+        this._send({ msg: 'pong', id: msg.id });
+      } else if (msg.msg === 'pong') {
+        // noop, as we assume everything's a pong
+      } else if (
+        ['added', 'changed', 'removed', 'ready', 'updated'].includes(msg.msg)
+      ) {
+        this._livedata_data(msg);
+      } else if (msg.msg === 'nosub') {
+        this._livedata_nosub(msg);
+      } else if (msg.msg === 'result') {
+        this._livedata_result(msg);
+      } else if (msg.msg === 'error') {
+        this._livedata_error(msg);
       } else {
-        const description =
-          'DDP version negotiation failed; server requested version ' +
-          msg.version;
-        this._stream.disconnect({ _permanent: true, _error: description });
-        this.options.onDDPVersionNegotiationFailure(description);
+        Meteor._debug('discarding unknown livedata message type', msg);
       }
-    } else if (msg.msg === 'ping' && this.options.respondToPings) {
-      this._send({ msg: 'pong', id: msg.id });
-    } else if (msg.msg === 'pong') {
-      // noop, as we assume everything's a pong
-    } else if (
-      ['added', 'changed', 'removed', 'ready', 'updated'].includes(msg.msg)
-    ) {
-      this._livedata_data(msg);
-    } else if (msg.msg === 'nosub') {
-      this._livedata_nosub(msg);
-    } else if (msg.msg === 'result') {
-      this._livedata_result(msg);
-    } else if (msg.msg === 'error') {
-      this._livedata_error(msg);
-    } else {
-      Meteor._debug('discarding unknown livedata message type', msg);
     }
   }
 

packages/ddp-common/utils.js

@@ -38,80 +38,118 @@ export function last(array, n, guard) {
   return slice.call(array, Math.max(array.length - n, 0));
 }
 
+DDPCommon.ALLOW_BUFFERING = false;
+
 DDPCommon.SUPPORTED_DDP_VERSIONS = [ '1', 'pre2', 'pre1' ];
 
 DDPCommon.parseDDP = function (stringMessage) {
   try {
-    var msg = JSON.parse(stringMessage);
+    var messages = JSON.parse(stringMessage);
   } catch (e) {
-    Meteor._debug("Discarding message with invalid JSON", stringMessage);
-    return null;
-  }
-  // DDP messages must be objects.
-  if (msg === null || typeof msg !== 'object') {
-    Meteor._debug("Discarding non-object DDP message", stringMessage);
+    Meteor._debug("Discarding message(s) with invalid JSON", stringMessage);
     return null;
   }
+
+  // Convert all DDP messages to Array form
+  messages = Array.isArray(messages) ? messages : [messages];
+
+  for (let i = 0; i < messages.length; i++) {
+    var msg = messages[i];
 
-  // massage msg to get it into "abstract ddp" rather than "wire ddp" format.
+    // Each individual DDP message must be an object.
+    if (msg === null || typeof msg !== 'object') {
+      Meteor._debug("Discarding non-object DDP message", stringMessage);
+
+      messages.splice(i, 1);
 
-  // switch between "cleared" rep of unsetting fields and "undefined"
-  // rep of same
-  if (hasOwn.call(msg, 'cleared')) {
-    if (! hasOwn.call(msg, 'fields')) {
-      msg.fields = {};
+      i--;
+
+      continue;
     }
-    msg.cleared.forEach(clearKey => {
-      msg.fields[clearKey] = undefined;
+
+    // massage msg to get it into "abstract ddp" rather than "wire ddp" format.
+
+    // switch between "cleared" rep of unsetting fields and "undefined"
+    // rep of same
+    if (hasOwn.call(msg, 'cleared')) {
+      if (! hasOwn.call(msg, 'fields')) {
+        msg.fields = {};
+      }
+      msg.cleared.forEach(clearKey => {
+        msg.fields[clearKey] = undefined;
+      });
+      delete msg.cleared;
+    }
+
+    ['fields', 'params', 'result'].forEach(field => {
+      if (hasOwn.call(msg, field)) {
+        msg[field] = EJSON._adjustTypesFromJSONValue(msg[field]);
+      }
     });
-    delete msg.cleared;
   }
 
-  ['fields', 'params', 'result'].forEach(field => {
-    if (hasOwn.call(msg, field)) {
-      msg[field] = EJSON._adjustTypesFromJSONValue(msg[field]);
-    }
-  });
+  const messagesLength = messages.length;
 
-  return msg;
+  if (messagesLength === 0) {
+    return null;
+  } else if (messagesLength === 1) {
+    return messages[0];
+  } else {
+    return messages;
+  }
 };
 
-DDPCommon.stringifyDDP = function (msg) {
-  const copy = EJSON.clone(msg);
+DDPCommon.stringifyDDP = function (messages) {
+  messages = Array.isArray(messages) ? messages : [messages];
+
+  const clonedMessages = [];
+
+  for (let i = 0; i < messages.length; i++) {
+    const msg = messages[i];
+    const copy = EJSON.clone(msg);
 
-  // swizzle 'changed' messages from 'fields undefined' rep to 'fields
-  // and cleared' rep
-  if (hasOwn.call(msg, 'fields')) {
-    const cleared = [];
+    // swizzle 'changed' messages from 'fields undefined' rep to 'fields
+    // and cleared' rep
+    if (hasOwn.call(msg, 'fields')) {
+      const cleared = [];
 
-    Object.keys(msg.fields).forEach(key => {
-      const value = msg.fields[key];
+      Object.keys(msg.fields).forEach(key => {
+        const value = msg.fields[key];
 
-      if (typeof value === "undefined") {
-        cleared.push(key);
-        delete copy.fields[key];
+        if (typeof value === "undefined") {
+          cleared.push(key);
+          delete copy.fields[key];
+        }
+      });
+
+      if (! isEmpty(cleared)) {
+        copy.cleared = cleared;
       }
-    });
 
-    if (! isEmpty(cleared)) {
-      copy.cleared = cleared;
+      if (isEmpty(copy.fields)) {
+        delete copy.fields;
+      }
     }
 
-    if (isEmpty(copy.fields)) {
-      delete copy.fields;
-    }
-  }
+    // adjust types to basic
+    ['fields', 'params', 'result'].forEach(field => {
+      if (hasOwn.call(copy, field)) {
+        copy[field] = EJSON._adjustTypesToJSONValue(copy[field]);
+      }
+    });
 
-  // adjust types to basic
-  ['fields', 'params', 'result'].forEach(field => {
-    if (hasOwn.call(copy, field)) {
-      copy[field] = EJSON._adjustTypesToJSONValue(copy[field]);
+    if (msg.id && typeof msg.id !== 'string') {
+      throw new Error("Message id is not a string");
     }
-  });
 
-  if (msg.id && typeof msg.id !== 'string') {
-    throw new Error("Message id is not a string");
+    clonedMessages.push(copy);
   }
 
-  return JSON.stringify(copy);
+  const messagesLength = clonedMessages.length;
+
+  if (messagesLength === 1) {
+    return JSON.stringify(clonedMessages[0]);
+  } else {
+    return JSON.stringify(clonedMessages);
+  }
 };