Extended types for DDP

This is a first pass for now; it doesn't support all the types
we will eventually support, and it may be in flux in terms of the exact
format for a little while yet.  Also I need to write tests.

But the outline is there.

packages/livedata/livedata_common.js

@@ -1,4 +1,4 @@
-// XXX namespacing
+(function () {
 Meteor._SUPPORTED_DDP_VERSIONS = [ 'pre1' ];
 
 Meteor._MethodInvocation = function (options) {
@@ -42,6 +42,157 @@ _.extend(Meteor._MethodInvocation.prototype, {
   }
 });
 
+
+var customTypes = {};
+// Add a custom type, using a method of your choice to get to and
+// from a basic JSON-able representation.
+// XXX: doc this
+Meteor.addCustomType = function (typeName, toBasic, fromBasic, recognize) {
+  if (_.has(customTypes), typeName)
+    throw new Error("Type " + typeName + " already present");
+  customTypes[typeName] = {toBasic: toBasic, fromBasic: fromBasic, recognize: recognize};
+};
+
+var builtinConverters = [
+  { // Date
+    matchBasic: function (obj) {
+      return _.has(obj, '$date') && _.size(obj) === 1;
+    },
+    matchObject: function (obj) {
+      return obj instanceof Date;
+    },
+    toBasic: function (obj) {
+      return {$date: obj.UTC()};
+    },
+    fromBasic: function (obj) {
+      return new Date(obj.$date);
+    }
+  },
+  { // Literal
+    matchBasic: function (obj) {
+      return _.has(obj, '$literal') && _.size(obj) === 1;
+    },
+    matchObject: function (obj) {
+      if (_.isEmpty(obj) || _.size(obj) > 2) {
+        return false;
+      }
+      return _.any(builtinConverters, function (converter) {
+        return converter.matchBasic(obj);
+      });
+    },
+    toBasic: function (obj) {
+      return {$literal: obj};
+    },
+    fromBasic: function (obj) {
+      return obj.$literal;
+    }
+  },
+  { // Custom
+    matchBasic: function (obj) {
+      return _.has(obj, '$type') && _.has(obj, '$value') && _.size(obj) === 2;
+    },
+    matchObject: function (obj) {
+      return _.any(customTypes, function (type) {
+        return type.recognize(obj);
+      });
+    },
+    toBasic: function (obj) {
+      var typeName = null;
+      var converter = _.find(customTypes, function(type, name) {
+        typeName = name;
+        return type.recognize(obj);
+      });
+      return {$type: typeName, $value: converter.toBasic(obj)};
+    },
+    fromBasic: function (obj) {
+      var converter = customTypes[obj.$type];
+      return converter.fromBasic(obj.$value);
+    }
+  }
+];
+
+
+//XXX: copypasta.  use string keys to control which functions
+//  I'm calling?
+var adjustTypesToBasic = function (obj) {
+  _.each(obj, function (value, key) {
+    if (typeof value !== 'object')
+      return; // continue
+    for (var i = 0; i < builtinConverters.length; i++) {
+      var converter = builtinConverters[i];
+      if (converter.matchObject(value)) {
+        obj[key] = converter.toBasic(value);
+        return; // continue to the next field
+      }
+    }
+    // if we get here, value is an object but not adjustable
+    // at this level.  recurse.
+    adjustTypesToBasic(value);
+  });
+};
+
+var adjustTypesFromBasic = function (obj) {
+  _.each(obj, function (value, key) {
+    if (typeof value !== 'object' || _.size(value) > 2)
+      return; // continue
+    for (var i = 0; i < builtinConverters.length; i++) {
+      var converter = builtinConverters[i];
+      if (converter.matchBasic(value)) {
+        obj[key] = converter.fromBasic(value);
+        return; // continue to the next field
+      }
+    }
+    // if we get here, value is an object but not adjustable
+    // at this level.  recurse.
+    adjustTypesFromBasic(value);
+  });
+};
+
+Meteor._parseDDP = function (stringMessage) {
+  var msg = JSON.parse(stringMessage);
+  //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 (_.has(msg, 'cleared')) {
+    if (!_.has(msg, 'fields'))
+      msg.fields = {};
+    _.each(msg.cleared, function (clearKey) {
+      msg.fields[clearKey] = undefined;
+    });
+    delete msg.cleared;
+  }
+
+  _.each(['fields', 'params'], function (field) {
+    if (_.has(msg, field))
+      adjustTypesFromBasic(msg[field]);
+  });
+  return msg;
+};
+
+Meteor._stringifyDDP = function (msg) {
+  var copy = LocalCollection._deepcopy(msg);
+  // swizzle 'changed' messages from 'fields undefined' rep to 'fields and cleared' rep
+  if (_.has(msg, 'fields')) {
+    var cleared = [];
+    _.each(msg.fields, function (value, key) {
+      if (key === undefined) {
+        cleared.push(key);
+        delete copy.fields[key];
+      }
+    });
+    if (!_.isEmpty(cleared))
+      copy.cleared = cleared;
+    if (_.isEmpty(copy.fields))
+      delete copy.fields;
+  }
+  // adjust types to basic
+  _.each(['fields', 'params'], function (field) {
+    if (_.has(copy, field))
+      adjustTypesToBasic(copy[field]);
+  });
+  return JSON.stringify(copy);
+};
+
 Meteor._CurrentInvocation = new Meteor.EnvironmentVariable;
 
 Meteor.Error = function (error, reason, details) {
@@ -66,3 +217,4 @@ Meteor.Error = function (error, reason, details) {
 };
 
 Meteor.Error.prototype = new Error;
+})();

packages/livedata/livedata_connection.js

@@ -4,20 +4,6 @@ if (Meteor.isServer) {
   var Future = __meteor_bootstrap__.require(path.join('fibers', 'future'));
 }
 
-var parseDDP = function (stringMessage) {
-  var msg = JSON.parse(stringMessage);
-  //massage msg to get it into "abstract ddp" rather than "wire ddp" format.
-  if (_.has(msg, 'cleared')) {
-    if (!_.has(msg, 'fields'))
-      msg.fields = {};
-    _.each(msg.cleared, function (clearKey) {
-      msg.fields[clearKey] = undefined;
-    });
-    delete msg.cleared;
-  }
-  return msg;
-};
-
 // @param url {String|Object} URL to Meteor app,
 //   or an object as a test hook (see code)
 // Options:
@@ -188,7 +174,7 @@ Meteor._LivedataConnection = function (url, options) {
 
   self._stream.on('message', function (raw_msg) {
     try {
-      var msg = parseDDP(raw_msg);
+      var msg = Meteor._parseDDP(raw_msg);
     } catch (err) {
       Meteor._debug("discarding message with invalid JSON", raw_msg);
       return;
@@ -750,7 +736,7 @@ _.extend(Meteor._LivedataConnection.prototype, {
   // Sends the DDP stringification of the given message object
   _send: function (obj) {
     var self = this;
-    self._stream.send(JSON.stringify(obj));
+    self._stream.send(Meteor._stringifyDDP(obj));
   },
 
   status: function (/*passthrough args*/) {
@@ -991,7 +977,7 @@ _.extend(Meteor._LivedataConnection.prototype, {
         throw new Error("It doesn't make sense to be adding something we know exists: "
                         + msg.id);
       }
-      serverDoc.document = msg.fields;
+      serverDoc.document = msg.fields || {};
       serverDoc.document._id = msg.id;
     } else {
       self._pushUpdate(updates, msg.collection, msg);

packages/livedata/livedata_server.js

@@ -278,23 +278,16 @@ _.extend(Meteor._LivedataSession.prototype, {
 
   sendChanged: function (collectionName, id, fields) {
     var self = this;
-    var cleared = [];
-    var messageFields = {};
     if (_.isEmpty(fields))
       return;
-    // convert internal format (undefined is clear) to wire format (list of clear)
-    _.each(fields, function (value, key) {
-      if (value === undefined)
-        cleared.push(key);
-      else
-        messageFields[key] = value;
-    });
+
     if (self._isSending) {
-      var toSend = {msg: "changed", collection: collectionName, id: id};
-      if (!_.isEmpty(messageFields))
-        toSend.fields = messageFields;
-      if (!_.isEmpty(cleared))
-        toSend.cleared = cleared;
+      var toSend = {
+        msg: "changed",
+        collection: collectionName,
+        id: id,
+        fields: fields
+      };
       self.send(toSend);
     }
   },
@@ -357,7 +350,7 @@ _.extend(Meteor._LivedataSession.prototype, {
     self.socket = socket;
     self.last_connect_time = +(new Date);
     _.each(self.out_queue, function (msg) {
-      self.socket.send(JSON.stringify(msg));
+      self.socket.send(Metoer._stringifyDDP(msg));
     });
     self.out_queue = [];
 
@@ -428,7 +421,7 @@ _.extend(Meteor._LivedataSession.prototype, {
   send: function (msg) {
     var self = this;
     if (self.socket)
-      self.socket.send(JSON.stringify(msg));
+      self.socket.send(Meteor._stringifyDDP(msg));
     else
       self.out_queue.push(msg);
   },
@@ -888,7 +881,7 @@ Meteor._LivedataServer = function () {
       var msg = {msg: 'error', reason: reason};
       if (offending_message)
         msg.offending_message = offending_message;
-      socket.send(JSON.stringify(msg));
+      socket.send(Meteor._stringifyDDP(msg));
     };
 
     socket.on('data', function (raw_msg) {
@@ -968,12 +961,12 @@ _.extend(Meteor._LivedataServer.prototype, {
       self.sessions[socket.meteor_session.id] = socket.meteor_session;
 
 
-      socket.send(JSON.stringify({msg: 'connected',
+      socket.send(Meteor._stringifyDDP({msg: 'connected',
                                   session: socket.meteor_session.id}));
       // will kick off previous connection, if any
       socket.meteor_session.connect(socket);
     } else {
-      socket.send(JSON.stringify({msg: 'failed', version: version}));
+      socket.send(Meteor._stringifyDDP({msg: 'failed', version: version}));
       socket.close();
     }
   },