MongoDB mixin: add `aggregate` to available methods

[PuKoren]

Hello there!

It would be super-great to be able to aggregate using the pump, sample:

pump
  .mixin(MongodbMixin('mongodb://localhost/myDB'))
  .useCollection('myCollection')
  .from(pump.aggregate([{ $group: { _id: '$geo.country', sum: { $sum: 1} }}, {$sort: { 'sum': -1}}]))

I don't know coffee, would it be hard to implement?

[rafaneri]

Hello,

I had try write a pump.aggregate without success. The collection.aggregate is called recursively, but I can't understood.

On the Pump, my buffer.on('data') is never called, and on the MongodbMixin target._mongo.collection.aggregate.apply(target._mongo.collection, args).stream().pipe(result) is called recursively.

I have used the _deferFind as base for my _deferAggregate

_wrapAggregate = (target) ->
  target.aggregate = (args...) ->
    if args instanceof Array
      args = args[0]
    throw new Error 'Collection is not set. Use .useCollection before using this wrapper' if !@_mongo.collection?
    throw new Error "Method .find() does not exist in collection." if !@_mongo.collection.find
    @_mongo.collection.aggregate.apply(@_mongo.collection, args).stream()

_deferAggregate = (collection, target) ->
  collection.aggregate = (args...) ->
    result = new PassThrough
      objectMode: true
      highWaterMark: 10

    target._mongo.whenConnected.then ->
      target._mongo.collection.aggregate.apply(target._mongo.collection, args).stream().pipe result

    { stream: -> result }

In _wrapAggregate I did an adjust in the arguments to get the first array index, because I send the aggregation pipeline as args, but they become a array of array through the function calls.

pump.aggregate([{aggregateQuery:...}])

my complete MongodbMixin.coffee implementation:

Promise = require 'bluebird'
mongo = Promise.promisifyAll require 'mongodb'
{ PassThrough } = require 'stream'

module.exports = (db) ->
  (target) ->
    target._mongo =
      db: db

    if typeof db == 'string'
      (target._mongo.whenConnected = mongo.MongoClient.connectAsync db)
        .then (db) ->
          target._mongo.db = db
          target.whenFinished().then -> target._mongo.db.close()
          if target._mongo?.collection?._datapumpsMixinName?
            target._mongo.collection = db.collection target._mongo.collection._datapumpsMixinName

    _wrapMethod target, name for name in _wrappedMethods
    _wrapFind target
    _wrapAggregate target

    target.setGlobalDefaultMaxBsonSize = (size) ->
     require('mongodb').Connection.DEFAULT_MAX_BSON_SIZE = size
     require('mongodb').Connection.DEFAULT_MAX_MESSAGE_SIZE = size
     @

    target.db = ->
      @_mongo.db

    # Use the given collection in the pump. The promisified methods of the collection will be proxied
    # from the pump, i.e. when you call .find() on the pump, it will call .find() on the collection.
    target.useCollection = (name) ->
      if target._mongo.whenConnected? and target._mongo?.whenConnected.isPending()
        @_mongo.collection =
          _datapumpsMixinName: name
        _deferCollectionMethod @_mongo.collection, name + 'Async', @ for name in _wrappedMethods
        _deferFind @_mongo.collection, @
        _deferAggregate @_mongo.collection, @
      else
        @_mongo.collection = @_mongo.db.collection name
      @

    # Returns a collection from the mongo database. Does not work until the connection established.
    target.collection = (name) ->
      throw new Error('Not yet connected to mongo') if target._mongo?.whenConnected.isPending()
      @_mongo.db.collection name

_wrappedMethods = [ 'insert', 'remove', 'rename', 'save', 'update', 'count', 'drop', 'findOne',
  'createIndex', 'ensureIndex', 'dropIndex', 'reIndex', 'group', 'options', 'indexes', 'stats',
  'findAndModify', 'findAndRemove' ]

_wrapMethod = (target, name) ->
  target[name] = (args...) ->
    throw new Error 'Collection is not set. Use .useCollection before using this wrapper' if !@_mongo.collection?
    throw new Error "Method .#{name}Async() does not exist in collection." if !@_mongo.collection[name + 'Async']
    @_mongo.collection[name + 'Async'].apply(@_mongo.collection, args)

_wrapFind = (target) ->
  target.find = (args...) ->
    throw new Error 'Collection is not set. Use .useCollection before using this wrapper' if !@_mongo.collection?
    throw new Error "Method .find() does not exist in collection." if !@_mongo.collection.find
    @_mongo.collection.find.apply(@_mongo.collection, args).stream()

_wrapAggregate = (target) ->
  target.aggregate = (args...) ->
    if args instanceof Array
      args = args[0]
    throw new Error 'Collection is not set. Use .useCollection before using this wrapper' if !@_mongo.collection?
    throw new Error "Method .aggregate() does not exist in collection." if !@_mongo.collection.aggregate
    @_mongo.collection.aggregate.apply(@_mongo.collection, args).stream()

_deferCollectionMethod = (collection, name, target) ->
  collection[name] = (args...)->
    target._mongo.whenConnected.then ->
      throw new Error "Method .#{name} does not exist in collection." if !target._mongo.collection[name]
      target._mongo.collection[name].apply(target._mongo.collection, args)

_deferFind = (collection, target) ->
  collection.find = (args...) ->
    result = new PassThrough
      objectMode: true
      highWaterMark: 10

    target._mongo.whenConnected.then ->
      target._mongo.collection.find.apply(target._mongo.collection, args).stream().pipe result

    { stream: -> result }

_deferAggregate = (collection, target) ->
  collection.aggregate = (args...) ->
    result = new PassThrough
      objectMode: true
      highWaterMark: 10

    target._mongo.whenConnected.then ->
      target._mongo.collection.aggregate.apply(target._mongo.collection, args).stream().pipe result

    { stream: -> result }

[novaki]

Hello @rafaneri

I'm sorry for the late reply. I've checked your code and I suspect that the recursion happens because the target._mongo.whenConnected is fulfilled, but the .then callback in _deferAggregrate is called before the .then callback of the mixin (line 12).

This could also happen with the .find method, so this may be a bug in my code. I would replace line 11 with this:

      target._mongo.whenConnected = mongo.MongoClient.connectAsync db

So this basically defers the fulfilment of whenConnected promise to the point where the deferred collection methods are overwritten with the real ones from the connection. Please check if it solves the problem.

From the coding point of view, I would not copy-paste the code, but create .deferCursorReturningMethod and .wrapCursorReturningMethod methods from the *find methods with an extra methodName argument, and use those to wrap and defer find and aggregate.