Aggregation (distinct, aggregate and mapReduce) extension for Meteor's Smart-collection

aggregation.js

var Future, path, _dummyCollection_, _futureWrapper;

path = Npm.require("path");
Future = Npm.require(path.join("fibers", "future"));
_dummyCollection_ = new Meteor.Collection('__dummy__');
_futureWrapper = function(collection, commandName, args) {
  var col, collectionName, future, result,
      _this = this;
  col = (typeof collection) === "string" ? _dummyCollection_ : collection;
  collectionName = (typeof collection) === "string" ? collection : collection._name;
  future = new Future;
  col.find()._mongo.db.collection(collectionName, function(err, collection) {
    if (err) {
      future["throw"](err);
    }
    return collection[commandName](args, function(err, result) {
      if (err) {
        future["throw"](err);
      }
      return future.ret([true, result]);
    });
  });
  result = future.wait();
  if (!result[0]) {
    throw result[1];
  }
  return result[1];
};

_callMapReduce = function(collection, map, reduce, options) {
    var col, collectionName, future, result;
    col = (typeof collection) === "string" ? _dummyCollection_ : collection;
    collectionName = (typeof collection) === "string" ? collection : collection._name;
    future = new Future;
    col.find()._mongo.db.collection(collectionName, function(err, coll) {
      if (err) {
        future["throw"](err);
      }
      return coll.mapReduce(map, reduce, options, function(err, result, stats) {
        var res;
        if (err) {
          future["throw"](err);
        }
        res = {
          collectionName: result.collectionName,
          stats: stats
        };
        return future.ret([true, res]);
      });
    });
    result = future.wait();
    if (!result[0]) {
      throw result[1];
    }
    return result[1];
  }

_.extend(Meteor.SmartCollection.prototype, {
  distinct: function(key) {
    return _futureWrapper(this._name, "distinct", key);
  },
  aggregate: function(pipeline) {
    return _futureWrapper(this._name, "aggregate", pipeline);
  },
  mapReduce: function(map, reduce, options) {
    return _callMapReduce(this._name, map, reduce, options);
  }
});