ReactiveAggregate returns Mongo types instead of Meteor

[abuszta]

Hello,

Because meteor-reactive-aggregate is using rawCollection to call aggregate on collection, result contains plain Mongo types instead of Meteor ones. It is a problem for custom EJSON types returned in aggregate or when ObjectID or Decimal128 is used (see Scale Factor aggregation in https://docs.mongodb.com/manual/tutorial/model-monetary-data/).

This can be fixed by adding same code which is used in meteor/mongo.js for converting types:

diff --git a/aggregate.js b/aggregate.js
index b7e7cc9..2cb5683 100644
--- a/aggregate.js
+++ b/aggregate.js
@@ -1,6 +1,81 @@
 import { Meteor } from 'meteor/meteor';
 import { Mongo } from 'meteor/mongo';
 
+// FIXME: Copied this from mongo.js
+import { NpmModuleMongodb as MongoDB } from 'meteor/npm-mongo';
+import { EJSON } from 'meteor/ejson';
+
+var replaceNames = function (filter, thing) {
+  if (typeof thing === "object" && thing !== null) {
+    if (_.isArray(thing)) {
+      return _.map(thing, _.bind(replaceNames, null, filter));
+    }
+
+    var ret = {};
+
+    _.each(thing, function (value, key) {
+      ret[filter(key)] = replaceNames(filter, value);
+    });
+
+    return ret;
+  }
+
+  return thing;
+};
+
+var unmakeMongoLegal = function (name) {
+  return name.substr(5);
+};
+
+var replaceMongoAtomWithMeteor = function (document) {
+  if (document instanceof MongoDB.Binary) {
+    var buffer = document.value(true);
+    return new Uint8Array(buffer);
+  }
+
+  if (document instanceof MongoDB.ObjectID) {
+    return new Mongo.ObjectID(document.toHexString());
+  }
+
+  if (document instanceof MongoDB.Decimal128) {
+    return Decimal(document.toString());
+  }
+
+  if (document["EJSON$type"] && document["EJSON$value"] && _.size(document) === 2) {
+    return EJSON.fromJSONValue(replaceNames(unmakeMongoLegal, document));
+  }
+
+  if (document instanceof MongoDB.Timestamp) {
+    // For now, the Meteor representation of a Mongo timestamp type (not a date!
+    // this is a weird internal thing used in the oplog!) is the same as the
+    // Mongo representation. We need to do this explicitly or else we would do a
+    // structural clone and lose the prototype.
+    return document;
+  }
+
+  return undefined;
+};
+
+var replaceTypes = function (document, atomTransformer) {
+  if (typeof document !== 'object' || document === null) return document;
+  var replacedTopLevelAtom = atomTransformer(document);
+  if (replacedTopLevelAtom !== undefined) return replacedTopLevelAtom;
+  var ret = document;
+
+  _.each(document, function (val, key) {
+    var valReplaced = replaceTypes(val, atomTransformer);
+
+    if (val !== valReplaced) {
+      // Lazy clone. Shallow copy.
+      if (ret === document) ret = _.clone(document);
+      ret[key] = valReplaced;
+    }
+  });
+
+  return ret;
+};
+//FIXME: END
+
 const defaultOptions = ({
   collection, options
 }) => ({
@@ -48,6 +123,7 @@ export const ReactiveAggregate = function (subscription, collection, pipeline =
       }
       // cursor is not done iterating, add and update documents on the client
       else {
+        doc = replaceTypes(doc, replaceMongoAtomWithMeteor);
         if (!subscription._ids[doc._id]) {
           subscription.added(clientCollection, doc._id, doc);
         } else {

I'm not sure however how to solve this without copy-paste.

Antoni.