add streaming api implementation for getEventsByRevision

Author: nanov

lib/databases/mongodb.js

@@ -2,12 +2,74 @@ var util = require('util'),
   Store = require('../base'),
   _ = require('lodash'),
   async = require('async'),
+  stream = require('stream'),
   mongo = Store.use('mongodb'),
   mongoVersion = Store.use('mongodb/package.json').version,
   isNew = mongoVersion.indexOf('1.') !== 0,
   ObjectID = isNew ? mongo.ObjectID : mongo.BSONPure.ObjectID,
   debug = require('debug')('eventstore:store:mongodb');
 
+function streamEventsByRevision(self, findStatement, revMin, revMax, resultStream, lastEvent) {
+
+  findStatement.streamRevision = (revMax === -1) ? { '$gte': revMin } : { '$gte': revMin, '$lt': revMax };
+
+  var mongoStream = self.events.find(findStatement, { sort: [['commitStamp', 'asc'], ['streamRevision', 'asc'], ['commitSequence', 'asc']] });
+
+  async.during(function(clb) {
+    mongoStream.hasNext(clb);
+  },
+  function(clb) {
+      mongoStream.next(function(error, e) {
+      if (error)
+        return clb(error);
+
+      if (!lastEvent) {
+        lastEvent = e;
+        return clb();
+      }
+
+      // if for some reason we have written this event alredy
+      if ((e.streamRevision === lastEvent.streamRevision && e.restInCommitStream <= lastEvent.restInCommitStream) ||
+          (e.streamRevision <= lastEvent.streamRevision)) {
+          return clb();
+      }
+
+      lastEvent = e;
+      resultStream.write(lastEvent, clb);
+    });
+  },
+  function (error) {
+    if (error) {
+      return resultStream.destroy(error);
+    }
+
+    if (!lastEvent) {
+      return resultStream.end();
+    }
+
+    var txOk = (revMax === -1 && !lastEvent.restInCommitStream) ||
+                (revMax !== -1 && (lastEvent.streamRevision === revMax - 1 || !lastEvent.restInCommitStream));
+
+    if (txOk) {
+      // the following is usually unnecessary
+      self.removeTransactions(lastEvent);
+      resultStream.end(lastEvent);
+    }
+
+    self.repairFailedTransaction(lastEvent, function (err) {
+      if (err) {
+        if (err.message.indexOf('missing tx entry') >= 0) {
+          return resultStream.end(lastEvent);;
+        }
+        debug(err);
+        return resultStream.destroy(error);
+      }
+
+      streamEventsByRevision(self, findStatement, lastEvent.revMin, revMax, resultStream, lastEvent);
+    });
+  });
+};
+  
 function Mongo(options) {
   options = options || {};
 
@@ -330,6 +392,33 @@ _.extend(Mongo.prototype, {
     return query;
   },
 
+  streamEventsByRevision: function (query, revMin, revMax) {
+    if (!query.aggregateId) {
+      var errMsg = 'aggregateId not defined!';
+      debug(errMsg);
+      if (callback) callback(new Error(errMsg));
+      return;
+    }
+
+    var findStatement = {
+      aggregateId: query.aggregateId,
+    };
+
+    if (query.aggregate) {
+      findStatement.aggregate = query.aggregate;
+    }
+
+    if (query.context) {
+      findStatement.context = query.context;
+    }
+
+    var self = this;
+
+    var resultStream = new stream.PassThrough({ objectMode: true, highWaterMark: 1 });
+    streamEventsByRevision(self, findStatement, revMin, revMax, resultStream);
+    return resultStream;
+  },
+
   getEvents: function (query, skip, limit, callback) {
     this.streamEvents(query, skip, limit).toArray(callback);
   },

lib/eventstore.js

@@ -143,8 +143,8 @@ _.extend(Eventstore.prototype, {
    * @param {Date}     commitStamp the date object
    * @param {Number}   skip        how many events should be skipped? [optional]
    * @param {Number}   limit       how many events do you want in the result? [optional]
-  * @returns {Stream} a stream with the events
-  */
+   * @returns {Stream} a stream with the events
+   */
   streamEventsSince: function (commitStamp, skip, limit) {
     if (!this.store.streamEvents) {
       throw new Error('Streaming API is not suppoted by '+(this.options.type || 'inmemory') +' db implementation.');
@@ -163,6 +163,28 @@ _.extend(Eventstore.prototype, {
   },
 
 
+  /**
+   * stream events by revision
+   * @param {Object || String} query    the query object
+   * @param {Number}           revMin   revision start point [optional]
+   * @param {Number}           revMax   revision end point (hint: -1 = to end) [optional]
+   * @returns {Stream} a stream with the events
+   */
+  streamEventsByRevision: function (query, revMin, revMax) {
+    if (typeof query === 'string') {
+      query = { aggregateId: query };
+    }
+
+    if (!query.aggregateId) {
+      var err = new Error('An aggregateId should be passed!');
+      debug(err);
+      if (callback) callback(err);
+      return;
+    }
+
+    return this.store.streamEventsByRevision(query, revMin, revMax);
+  },
+
   /**
    * loads the events
    * @param {Object || String} query    the query object [optional]