[WIP] Implement batching all the way from MongoDB => DDP

packages/ddp-client/common/livedata_connection.js

@@ -451,7 +451,12 @@ export class Connection {
           }
         }
       };
-      self._send({ msg: 'sub', id: id, name: name, params: params });
+      self._send({
+        msg: 'sub',
+        id: id,
+        name: name,
+        params: params,
+      });
     }
 
     // return a handle to the application.
@@ -794,13 +799,16 @@ export class Connection {
     }
 
     // If we added it to the first block, send it out now.
-    if (self._outstandingMethodBlocks.length === 1) methodInvoker.sendMessage();
+    if (self._outstandingMethodBlocks.length === 1) {
+      methodInvoker.sendMessage();
+    }
 
     // If we're using the default callback on the server,
     // block waiting for the result.
     if (future) {
       return future.wait();
     }
+
     return options.returnStubValue ? stubReturnValue : undefined;
   }
 
@@ -1561,9 +1569,28 @@ export class Connection {
       return;
     }
 
+    var messages = [];
+
     self._outstandingMethodBlocks[0].methods.forEach(m => {
-      m.sendMessage();
+      if (m.gotResult()) {
+        throw new Error('sendingMethod is called on method with result');
+      } 
+
+      // If we're re-sending it, it doesn't matter if data was written the first
+      // time.
+      m._dataVisible = false;
+      m.sentMessage = true;
+
+      // If m is a wait method, make all data messages be buffered until it is
+      // done.
+      if (m._wait) {
+        self.connection._methodsBlockingQuiescence[this.methodId] = true;
+      }
+
+      messages.push(m._message);
     });
+
+    self._send(messages);
   }
 
   _livedata_error(msg) {
@@ -1633,58 +1660,68 @@ export class Connection {
 
   onMessage(raw_msg) {
     try {
-      var msg = DDPCommon.parseDDP(raw_msg);
+      var messages = DDPCommon.parseDDP(raw_msg);
     } catch (e) {
       Meteor._debug('Exception while parsing DDP', e);
       return;
     }
 
+    // TODO: remove this console.log, which clearly shows the effect
+    console.log(messages.length);
+
     // Any message counts as receiving a pong, as it demonstrates that
     // the server is still alive.
     if (this._heartbeat) {
       this._heartbeat.messageReceived();
     }
 
-    if (msg === null || !msg.msg) {
-      // XXX COMPAT WITH 0.6.6. ignore the old welcome message for back
-      // compat.  Remove this 'if' once the server stops sending welcome
-      // messages (stream_server.js).
-      if (!(msg && msg.server_id))
-        Meteor._debug('discarding invalid livedata message', msg);
-      return;
-    }
+    // Convert all DDP messages to Array form
+    messages = Array.isArray(messages) ? messages : [messages];
+
+    for (var i = 0; i < messages.length; i++) {
+      var msg = messages[i];
+
+      if (msg === null || !msg.msg) {
+        // XXX COMPAT WITH 0.6.6. ignore the old welcome message for back
+        // compat.  Remove this 'if' once the server stops sending welcome
+        // messages (stream_server.js).
+        if (!(msg && msg.server_id))
+          Meteor._debug('discarding invalid livedata message', msg);
+        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 {
+          var 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 {
-        var 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);
     }
   }
 
@@ -1771,7 +1808,8 @@ export class Connection {
         msg: 'sub',
         id: id,
         name: sub.name,
-        params: sub.params
+        params: sub.params,
+        allowBatching: true,
       });
     });
   }

packages/ddp-common/utils.js

@@ -38,80 +38,96 @@ export function last(array, n, guard) {
   return slice.call(array, Math.max(array.length - n, 0));
 }
 
-DDPCommon.SUPPORTED_DDP_VERSIONS = [ '1', 'pre2', 'pre1' ];
+DDPCommon.SUPPORTED_DDP_VERSIONS = [ '2', '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);
-    return null;
-  }
 
-  // massage msg to get it into "abstract ddp" rather than "wire ddp" format.
+  // 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];
 
-  // switch between "cleared" rep of unsetting fields and "undefined"
-  // rep of same
-  if (hasOwn.call(msg, 'cleared')) {
-    if (! hasOwn.call(msg, 'fields')) {
-      msg.fields = {};
+    // Each individual DDP message must be an object.
+    if (msg === null || typeof msg !== 'object') {
+      Meteor._debug("Discarding non-object DDP message", stringMessage);
+      return null;
     }
-    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]);
+    // 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;
     }
-  });
 
-  return msg;
+    ['fields', 'params', 'result'].forEach(field => {
+      if (hasOwn.call(msg, field)) {
+        msg[field] = EJSON._adjustTypesFromJSONValue(msg[field]);
+      }
+    });
+  }
+
+  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];
 
-  // 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 msg.fields[key];
+        }
+      });
+
+      if (! isEmpty(cleared)) {
+        msg.cleared = cleared;
       }
-    });
 
-    if (! isEmpty(cleared)) {
-      copy.cleared = cleared;
+      if (isEmpty(msg.fields)) {
+        delete msg.fields;
+      }
     }
 
-    if (isEmpty(copy.fields)) {
-      delete copy.fields;
-    }
-  }
+    // adjust types to basic
+    ['fields', 'params', 'result'].forEach(field => {
+      if (hasOwn.call(msg, field)) {
+        msg[field] = EJSON._adjustTypesToJSONValue(msg[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(EJSON.clone(msg));
   }
 
-  return JSON.stringify(copy);
+  return JSON.stringify(clonedMessages);
 };

packages/ddp-server/common.js

@@ -0,0 +1,10 @@
+// Audit argument checks, if the audit-argument-checks package exists (it is a
+// weak dependency of this package).
+export function maybeAuditArgumentChecks(f, context, args, description) {
+  args = args || [];
+  if (Package['audit-argument-checks']) {
+    return Match._failIfArgumentsAreNotAllChecked(
+      f, context, args, description);
+  }
+  return f.apply(context, args);
+}