cqrs-eventdenormalizer - npm Package Compare versions

Comparing version 0.4.1 to 1.0.0

index.js

@@ -1,13 +0,36 @@
-var index;
+'use strict';
 
-if (typeof module.exports !== 'undefined') {
-    index = module.exports;
-} else {
-    index = root.index = {};
+var Denormalizer = require('./lib/denormalizer'),
+  _ = require('lodash'),
+  fs = require('fs'),
+  path = require('path');
+
+function denormalizer (options) {
+  return new Denormalizer(options);
 }
 
-index.VERSION = '0.0.1';
+/**
+ * Calls the constructor.
+ * @param  {Object} klass Constructor function.
+ * @param  {Array}  args  Arguments for the constructor function.
+ * @return {Object}       The new object.
+ */
+function construct(klass, args) {
+  function T() {
+    klass.apply(this, arguments[0]);
+  }
+  T.prototype = klass.prototype;
+  return new T(args);
+}
 
-index.eventDenormalizer = require('./lib/eventDenormalizer');
-index.viewBuilderBase = require('./lib/bases/viewBuilderBase');
-index.eventExtenderBase = require('./lib/bases/eventExtenderBase');
+var files = fs.readdirSync(path.join(__dirname, 'lib/definitions'));
+
+files.forEach(function (file) {
+  var name = path.basename(file, '.js');
+  var nameCap = name.charAt(0).toUpperCase() + name.slice(1);
+  denormalizer['define' + nameCap] = function () {
+    return construct(require('./lib/definitions/' + name), _.toArray(arguments));
+  };
+});
+
+module.exports = denormalizer;

lib/eventDispatcher.js

@@ -1,123 +0,137 @@
-var async = require('async')
-  , _ = require('lodash')
-  , eventEmitter = require('./eventEmitter')
-  , selfExtendHandle;
+'use strict';
 
-module.exports = {
+var debug = require('debug')('denormalizer:eventDispatcher'),
+  _ = require('lodash'),
+  async = require('async'),
+  dotty = require('dotty');
 
-    configure: function(fn) {
-        fn.call(this);
-        return this;
-    },
+/**
+ * EventDispatcher constructor
+ * @param {Object} tree       The tree object.
+ * @param {Object} definition The definition object.
+ * @constructor
+ */
+function EventDispatcher (tree, definition) {
+  if (!tree || !_.isObject(tree) || !_.isFunction(tree.getViewBuilders)) {
+    var err = new Error('Please pass a valid tree!');
+    debug(err);
+    throw err;
+  }
 
-    use: function(module) {
-        if (!module) return;
-    
-        if (module.push) {
-            this.eventQueue = module;
-        }
-    },
+  if (!definition || !_.isObject(definition)) {
+    var err = new Error('Please pass a valid command definition!');
+    debug(err);
+    throw err;
+  }
 
-    initialize: function(options, callback) {
-        var self = this;
+  this.tree = tree;
 
-        if (!callback) {
-            callback = options;
-            options = {};
-        }
+  this.definition = {
+    correlationId: 'correlationId', // optional
+    id: 'id',                       // optional
+    name: 'name',                   // optional
+//      aggregateId: 'aggregate.id',    // optional
+//      context: 'context.name',        // optional
+//      aggregate: 'aggregate.name',    // optional
+    payload: 'payload'              // optional
+//      revision: 'revision'            // optional
+//      version: 'version',             // optional
+//      meta: 'meta'                    // optional, if defined theses values will be copied to the notification (can be used to transport information like userId, etc..)
+  };
 
-        this.options = options ||  {};
+  this.definition = _.defaults(definition, this.definition);
+}
 
-        eventEmitter.on('extend:*', selfExtendHandle = function(evt) {
-            // var listeners = _.filter(eventEmitter.listeners('extend:' + evt.event), function(listener) {
-            //     return listener !== selfExtendHandle;
-            // });
+EventDispatcher.prototype = {
 
-            // if (listeners.length !== 1) {
-            if (eventEmitter.registerCount('extend:' + evt.event) === 0) {
-                eventEmitter.emit('extended:' + evt.event, evt);
-            }
-        });
+  /**
+   * Returns the target information of this event.
+   * @param {Object} evt The passed event.
+   * @returns {{name: 'eventName', aggregateId: 'aggregateId', version: 0, aggregate: 'aggregateName', context: 'contextName'}}
+   */
+  getTargetInformation: function (evt) {
+    if (!evt || !_.isObject(evt)) {
+      var err = new Error('Please pass a valid event!');
+      debug(err);
+      throw err;
+    }
 
-        eventEmitter.on('denormalized:*', function(evt) {
-            if (self.eventQueue) {
-                self.eventQueue.decrement(evt.id, function(err, removed) {
-                    if (removed) {
-                        eventEmitter.emit('extend:' + evt.event, evt);
-                        eventEmitter.emit('finishedDenormalization:' + evt.event + ':' + evt.id, evt);
-                    }
-                });
-            } else {
-                eventEmitter.emit('extend:' + evt.event, evt);
-                eventEmitter.emit('finishedDenormalization:' + evt.event + ':' + evt.id, evt);
-            }
-        });
+    var name = dotty.get(evt, this.definition.name) || '';
 
-        if (this.eventQueue) {
-            this.resetWorkers(function(err) {
-                self.reEmitEvents(callback);
-            });
-        } else {
-            callback(null);
-        }
-    },
+    var version = 0;
+    if (dotty.exists(evt, this.definition.version)) {
+      version = dotty.get(evt, this.definition.version);
+    } else {
+      debug('no version found, handling as version: 0');
+    }
 
-    resetWorkers: function(callback) {
-        var self = this;
+    var aggregate = null;
+    if (dotty.exists(evt, this.definition.aggregate)) {
+      aggregate = dotty.get(evt, this.definition.aggregate);
+    } else {
+      debug('no aggregate found');
+    }
 
-        if (this.eventQueue) {
-            this.eventQueue.getAll(function(err, items) {
-                async.forEach(items, function(item, cb) {
-                    // item.data.workers = eventEmitter.listeners('denormalize:' + item.data.event.event).length;
-                    item.data.workers = eventEmitter.registerCount('denormalize:' + item.data.event.event);
-                    self.eventQueue.push(item.id, item.data, cb);
-                }, callback);
-            });
-        } else {
-            callback(null);
-        }
-    },
+    var context = null;
+    if (dotty.exists(evt, this.definition.context)) {
+      context = dotty.get(evt, this.definition.context);
+    } else {
+      debug('no aggregateName found');
+    }
 
-    reEmitEvents: function(callback) {
-        if (this.eventQueue) {
-            this.eventQueue.getAll(function(err, items) {
-                async.forEach(items, function(item, cb) {
-                    eventEmitter.emit('denormalize:' + item.data.event.event, item.data.event);
-                    cb();
-                }, callback);
-            });
-        } else {
-            callback(null);
-        }
-    },
+    return {
+      name: name,
+      version: version,
+      aggregate: aggregate,
+      context: context
+    };
+  },
 
-    dispatch: function(evt, callback) {
+  /**
+   * Dispatches an event.
+   * @param {Object}   evt      The passed event.
+   * @param {Function} callback The function, that will be called when this action is completed.
+   *                            `function(errs, notifications){}`
+   */
+  dispatch: function (evt, callback) {
+    if (!evt || !_.isObject(evt)) {
+      var err = new Error('Please pass a valid event!');
+      debug(err);
+      throw err;
+    }
 
-        var entry = {
-            // workers: eventEmitter.listeners('denormalize:' + evt.event).length,
-            workers: eventEmitter.registerCount('denormalize:' + evt.event),
-            event: evt
-        };
+    if (!callback || !_.isFunction(callback)) {
+      var err = new Error('Please pass a valid callback!');
+      debug(err);
+      throw err;
+    }
 
-        // var extendersCount = _.filter(eventEmitter.listeners('extend:' + evt.event), function(listener) {
-        //     return listener !== selfExtendHandle;
-        // }).length;
+    var target = this.getTargetInformation(evt);
+    
+    var viewBuilders = this.tree.getViewBuilders(target);
 
-        var extendersCount = eventEmitter.registerCount('extend:' + evt.event);
-        if (entry.workers === 0 && extendersCount === 0) {
-            eventEmitter.emit('extended:' + evt.event, evt);
-            if (callback) callback(null);
-            return;
+    var errs = [];
+    var notifications = [];
+    async.each(viewBuilders, function (viewBuilder, callback) {
+      viewBuilder.denormalize(evt, function (err, notificaiton) {
+        if (err) {
+          debug(err);
+          errs.push(err);
         }
 
-        if (entry.workers === 0 && extendersCount > 0) {
-            eventEmitter.emit('extend:' + evt.event, evt);
-            if (callback) callback(null);
-            return;
+        if (notificaiton) {
+          notifications.push(notificaiton);
         }
+        callback(null);
+      })
+    }, function () {
+      if (errs.length === 0) {
+        errs = null;
+      }
+      callback(errs, notifications);
+    });
+  }
 
-        eventEmitter.emit('denormalize:' + entry.event.event, entry.event);
-        if (callback) callback(null);
-    }
+};
 
-};
+module.exports = EventDispatcher;

lib/orderQueue.js

@@ -1,24 +0,54 @@
-var _ = require('lodash');
+'use strict';
 
-var Queue = function(options) {
+var debug = require('debug')('denormalizer:orderQueue'),
+  _ = require('lodash');
+
+/**
+ * Queue constructor
+ * @param {Object} options The options object. Like: { queueTimeout: 3000 }
+ * @constructor
+ */
+function Queue (options) {
   this.queue = {};
   this.retries = {};
   this.options = options || { queueTimeout: 3000 };
-};
+}
 
 Queue.prototype = {
 
-  push: function(id, object, fn) {
-    if(!this.queue[id]) this.queue[id] = [];
-    this.queue[id].push(object);
-    if(!this.retries[id]) this.retries[id] = {};
-    if (object.id) {
-      this.retries[id][object.id] = this.retries[id][object.id] || 0;
+  /**
+   * Pushes a new item in the queue.
+   * @param {String}   id     The aggregate id.
+   * @param {String}   objId  The event id.
+   * @param {Object}   object The event.
+   * @param {Function} clb    The callback function for the event handle. 
+   * @param {Function} fn     The timeout function handle.
+   */
+  push: function (id, objId, object, clb, fn) {
+    this.queue[id] = this.queue[id] || [];
+    
+    var alreadyInQueue = _.find(this.queue[id], function (o) {
+      return o.id === objId && o.payload === object;
+    });
+    
+    if (alreadyInQueue) {
+      return;
     }
+    
+    this.queue[id].push({ id: objId, payload: object, callback: clb });
+    
+    this.retries[id] = this.retries[id] || {};
+    
+    this.retries[id][objId] = this.retries[id][objId] || 0;
+    
     if (fn) {
       var self = this;
-      (function wait() {
-        setTimeout(function() {
-          if (_.indexOf(self.queue[id], object) >= 0) {
-            var loopCount = self.retries[id][object.id]++;
+      (function wait () {
+        debug('wait called');
+        setTimeout(function () {
+          var found = _.find(self.queue[id], function (o) {
+            return o.id === objId;
+          });
+          if (found) {
+            var loopCount = self.retries[id][objId]++;
             fn(loopCount, wait);
@@ -31,18 +61,38 @@ }
 
-  get: function(id) {
+  /**
+   * Returns the pending events for an aggregate.
+   * @param {String} id The aggregate id.
+   * @returns {Array}
+   */
+  get: function (id) {
+    if (!this.queue[id] || this.queue[id].length === 0) {
+      return null;
+    }
     return this.queue[id];
   },
 
-  remove: function(id, object) {
+  /**
+   * Removes an event from the queue.
+   * @param {String} id    The aggregate id.
+   * @param {String} objId The event id.
+   */
+  remove: function (id, objId) {
     if (this.queue[id]) {
-      this.queue[id].splice(_.indexOf(this.queue[id], object), 1);
+      _.remove(this.queue[id], function (o) {
+        return o.id === objId;
+      });
     }
 
-    if (object.id && this.retries[id] && this.retries[id][object.id]) {
-      this.retries[id][object.id] = 0;
+    if (objId && this.retries[id] && this.retries[id][objId]) {
+      this.retries[id][objId] = 0;
     }
   },
 
+  /**
+   * NEVER USE THIS FUNCTION!!! ONLY FOR TESTS!
+   * clears the complete queue...
+   */
   clear: function() {
     this.queue = {};
+    this.retries = {};
   }
@@ -52,2 +102,2 @@
 
-module.exports = Queue;
+module.exports = Queue;

lib/replayHandler.js

@@ -1,36 +0,92 @@
-var _ = require('lodash'),
-		async = require('async'),
-		viewBuilders,
-		guardStore;
+'use strict';
 
-module.exports = {
-	initialize: function(options, vB, gS) {
-		if (!gS) {
-			gS = vB;
-			vB = options;
-			options = {};
-		}
+var debug = require('debug')('denormalizer:replayHandler'),
+  _ = require('lodash'),
+  async = require('async'),
+  dotty = require('dotty');
 
-		viewBuilders = vB;
+/**
+ * ReplayHandler constructor
+ * @param {Object} disp    The dispatcher object.
+ * @param {Object} store   The store object.
+ * @param {Object} def     The definition object.
+ * @param {Object} options The options object.
+ * @constructor
+ */
+function ReplayHandler (disp, store, def, options) {
+  this.dispatcher = disp;
 
-		guardStore = gS;
-	},
+  this.store = store;
+  
+  def = def || {};
 
-	replay: function(evts, callback) {
+  this.definition = {
+    correlationId: 'correlationId', // optional
+    id: 'id',                       // optional
+    name: 'name',                   // optional
+//      aggregateId: 'aggregate.id',    // optional
+//      context: 'context.name',        // optional
+//      aggregate: 'aggregate.name',    // optional
+    payload: 'payload'              // optional
+//      revision: 'revision'            // optional
+//      version: 'version',             // optional
+//      meta: 'meta'                    // optional, if defined theses values will be copied to the notification (can be used to transport information like userId, etc..)
+  };
 
-    var revisionMap = {},
-        groupedEvents = {};
+  this.definition = _.defaults(def, this.definition);
 
-    _.each(evts, function(evt) {
-      if (evt.head && evt.head.revision) {
-        revisionMap[evt.payload.id] = evt.head.revision;
-      }
+  options = options || {};
 
-      var interested = _.filter(viewBuilders, function(vB) {
-        return _.contains(vB.registeredEventNames, evt.event);
+  this.options = options;
+}
+
+ReplayHandler.prototype = {
+
+  /**
+   * Updates the revision in the store.
+   * @param {Object}   revisionMap The revision map.
+   * @param {Function} callback    The function, that will be called when this action is completed.
+   *                               `function(err){}`
+   */
+  updateRevision: function (revisionMap, callback) {
+    var self = this;
+    var ids = _.keys(revisionMap);
+    async.each(ids, function (id, callback) {
+      self.store.get(id, function (err, rev) {
+        if (err) {
+          return callback(err);
+        }
+        self.store.set(id, revisionMap[id] + 1, rev, callback);
       });
+    }, callback);
+  },
 
-      _.each(interested, function(inter) {
-        groupedEvents[inter.id] = groupedEvents[inter.id] || [];
-        groupedEvents[inter.id].push(evt);
+  /**
+   * Replays all passed events.
+   * @param {Array}    evts     The passed array of events.
+   * @param {Function} callback The function, that will be called when this action is completed.
+   *                            `function(err){}`
+   */
+  replay: function (evts, callback) {
+    var self = this;
+    
+    var revisionMap = {};
+    var viewBuilderMap = {};
+    var groupedEvents = {};
+
+    _.each(evts, function (evt) {
+      if (!!self.definition.revision && dotty.exists(evt, self.definition.revision) &&
+        !!self.definition.aggregateId && dotty.exists(evt, self.definition.aggregateId)) {
+        var aggId = dotty.get(evt, self.definition.aggregateId);
+        revisionMap[aggId] = dotty.get(evt, self.definition.revision);
+      }
+
+      var target = self.dispatcher.getTargetInformation(evt);
+
+      var viewBuilders = self.dispatcher.tree.getViewBuilders(target);
+
+      _.each(viewBuilders, function (vb) {
+        viewBuilderMap[vb.workerId] = vb;
+        groupedEvents[vb.workerId] = groupedEvents[vb.workerId] || [];
+        groupedEvents[vb.workerId].push(evt);
       });
@@ -40,76 +96,81 @@ });
     async.series([
-      function(callback) {
-        async.each(viewBuilders, function(viewBuilder, callback) {
-          if (!groupedEvents[viewBuilder.id] || groupedEvents[viewBuilder.id].length === 0) {
+      
+      function (callback) {
+        async.each(_.values(viewBuilderMap), function (vb, callback) {
+          if (!groupedEvents[vb.workerId] || groupedEvents[vb.workerId].length === 0) {
             return callback(null);
           }
 
-          viewBuilder.replay(groupedEvents[viewBuilder.id], callback);
-
+          vb.replay(groupedEvents[vb.workerId], callback);
         }, callback);
       },
-      function(callback) {
-        var ids = _.keys(revisionMap);
-        async.each(ids, function(id, callback) {
-          guardStore.getRevision(id, function(err, entry) {
-            if (err) { return callback(err); }
-
-            entry.revision = revisionMap[id] + 1;
-            guardStore.saveRevision(entry, callback);
-          });
-        }, callback);
+      
+      function (callback) {
+        self.updateRevision(revisionMap, callback);
       }
-    ], function(err) {
-      if (callback) callback(err);
+      
+    ], function (err) {
+      if (err) {
+        debug(err);
+      }
+      if (callback) {
+        callback(err);
+      }
     });
   },
 
-  replayStreamed: function(fn, retryTimout) {
+  /**
+   * Replays in a streamed way.
+   * @param {Function} fn The function that will be called with the replay function and the done function.
+   *                      `function(replay, done){}`
+   */
+  replayStreamed: function (fn) {
+    var self = this;
+    
+    var revisionMap = {};
+    var vbReplayStreams = {};
 
-    var revisionMap = {},
-        viewBuilderReplayStreams = {};
-
-    var replay = function(evt) {
-      if (evt.head && evt.head.revision) {
-        revisionMap[evt.payload.id] = evt.head.revision;
+    var replay = function (evt) {
+      if (!!self.definition.revision && dotty.exists(evt, self.definition.revision) &&
+          !!self.definition.aggregateId && dotty.exists(evt, self.definition.aggregateId)) {
+        var aggId = dotty.get(evt, self.definition.aggregateId);
+        revisionMap[aggId] = dotty.get(evt, self.definition.revision);
       }
 
-      var interested = _.filter(viewBuilders, function(vB) {
-        return _.contains(vB.registeredEventNames, evt.event);
-      });
+      var target = self.dispatcher.getTargetInformation(evt);
 
-      _.each(interested, function(inter) {
-        if (!viewBuilderReplayStreams[inter.id]) {
-          inter.replayStreamed(function(vbReplay, vbDone) {
-            viewBuilderReplayStreams[inter.id] = {
+      var viewBuilders = self.dispatcher.tree.getViewBuilders(target);
+
+      _.each(viewBuilders, function (vb) {
+
+        if (!vbReplayStreams[vb.workerId]) {
+          vb.replayStreamed(function (vbReplay, vbDone) {
+            vbReplayStreams[vb.workerId] = {
               replay: vbReplay,
               done: vbDone
             };
-          }, retryTimout);
+          });
         }
 
-        viewBuilderReplayStreams[inter.id].replay(evt);
+        vbReplayStreams[vb.workerId].replay(evt);
       });
     };
 
-    var done = function(callback) {
+    var done = function (callback) {
       async.series([
-        function(callback) {
-          async.each(_.values(viewBuilderReplayStreams), function(viewBuilderReplayStream, callback) {
-            viewBuilderReplayStream.done(callback);
+        function (callback) {
+          async.each(_.values(vbReplayStreams), function (vbReplayStream, callback) {
+            vbReplayStream.done(callback);
           }, callback);
         },
-        function(callback) {
-          var ids = _.keys(revisionMap);
-          async.each(ids, function(id, callback) {
-            guardStore.getRevision(id, function(err, entry) {
-              if (err) { return callback(err); }
-
-              entry.revision = revisionMap[id] + 1;
-              guardStore.saveRevision(entry, callback);
-            });
-          }, callback);
+        function (callback) {
+          self.updateRevision(revisionMap, callback);
         }
       ], function(err) {
-        if (callback) callback(err);
+        if (err) {
+          debug(err);
+        }
+        if (callback) {
+          callback(err);
+        }
       });
@@ -120,2 +181,5 @@ };
   }
-};
+
+};
+
+module.exports = ReplayHandler;

lib/revisionGuard.js

@@ -1,236 +0,287 @@
-var _ = require('lodash'),
-    Queue = require('./orderQueue'),
-    eventEmitter = require('./eventEmitter');
+'use strict';
 
-function randomBetween(min, max) {
-  return Math.round(min + Math.random() * (max - min));
-}
+var debug = require('debug')('denormalizer:revisionGuard'),
+  _ = require('lodash'),
+  async = require('async'),
+  Queue = require('./orderQueue'),
+  ConcurrencyError = require('./errors/concurrencyError'),
+  AlreadyDenormalizedError = require('./errors/alreadyDenormalizedError'),
+  dotty = require('dotty');
 
-module.exports = {
+/**
+ * RevisionGuard constructor
+ * @param {Object} store   The store object.
+ * @param {Object} options The options object.
+ * @constructor
+ */
+function RevisionGuard (store, options) {
+  options = options || {};
 
-  _getAux: function() {
-    var self = this;
+  var defaults = {
+    queueTimeout: 1000,
+    queueTimeoutMaxLoops: 3
+  };
+  
+  _.defaults(options, defaults);
+  
+  this.options = options;
 
-    this._aux = this._aux || {
+  if (!store || !_.isObject(store)) {
+    var err = new Error('store not injected!');
+    debug(err);
+    throw err;
+  }
+  
+  this.store = store;
 
-      finishGuard: function(evt, entry, callback) {
-        var aux = self._getAux();
+  this.definition = {
+    correlationId: 'correlationId', // optional
+    id: 'id',                       // optional
+    name: 'name',                   // optional
+//    aggregateId: 'aggregate.id',    // optional
+//    context: 'context.name',        // optional
+//    aggregate: 'aggregate.name',    // optional
+    payload: 'payload'              // optional
+//    revision: 'revision'            // optional
+//    version: 'version',             // optional
+//    meta: 'meta'                    // optional, if defined theses values will be copied to the notification (can be used to transport information like userId, etc..)
+  };
 
-        function finish() {
-          aux.revisionUpdateStrategy(entry, evt);
+  this.queue = new Queue({ queueTimeout: this.options.queueTimeout });
 
-          self.store.saveRevision(entry, function(err) {
-            if (err && err.name === 'ConcurrencyError') {
-              setTimeout(function() {
-                aux.retryGuard(evt);
-              }, 100);
-              return;
-            }
-            if (err) {
-              throw err;
-            }
+  this.onEventMissing(function (info, evt) {
+    debug('missing events: ', info, evt);
+  });
+}
 
-            aux.dequeuingStrategy(entry);
-          });
-        }
+/**
+ * Returns a random number between passed values of min and max.
+ * @param {Number} min The minimum value of the resulting random number.
+ * @param {Number} max The maximum value of the resulting random number.
+ * @returns {Number}
+ */
+function randomBetween(min, max) {
+  return Math.round(min + Math.random() * (max - min));
+}
 
-        // if (eventEmitter.listeners('denormalize:' + evt.event).length === 0) {
-        if (eventEmitter.registerCount('denormalize:' + evt.event) === 0) {
-          finish();
-        } else {
-          eventEmitter.once('finishedDenormalization:' + evt.event + ':' + evt.id, finish);
-        }
+RevisionGuard.prototype = {
+  
+  /**
+   * Inject definition for event structure.
+   * @param   {Object} definition the definition to be injected
+   */
+  defineEvent: function (definition) {
+    if (!_.isObject(definition)) {
+      throw new Error('Please pass in an object');
+    }
+    this.definition = _.defaults(definition, this.definition);
+    return this;
+  },
 
-        self.eventDispatcher.dispatch(evt, callback);
-      },
+  /**
+   * Inject function for event missing handle.
+   * @param   {Function} fn       the function to be injected
+   * @returns {RevisionGuard} to be able to chain...
+   */
+  onEventMissing: function (fn) {
+    if (!fn || !_.isFunction(fn)) {
+      var err = new Error('Please pass a valid function!');
+      debug(err);
+      throw err;
+    }
 
-      retryGuard: function(evt, callback) {
-        var aux = self._getAux();
-        self.store.getRevision(evt.payload.id, function(err, entry) {
-          if (aux.wouldQueue(evt, entry)) {
-            if (callback) callback(true, entry);
-            return;
-          }
+    if (fn.length === 1) {
+      fn = _.wrap(fn, function(func, info, evt, callback) {
+        func(info, evt);
+        callback(null);
+      });
+    }
 
-          aux.finishGuard(evt, entry, callback);
-        });
-      },
+    this.onEventMissingHandle = fn;
 
-      queueEvent: function(id, evt) {
-        var aux = self._getAux();
+    return this;
+  },
 
-        self.queue.push(id, evt, function(loopCount, waitAgain) {
-          aux.retryGuard(evt, function(err, entry) {
-            if (err) {
-              if (loopCount < self.options.queueTimeoutMaxLoops) {
-                return waitAgain();
-              }
-              // try to replay depending from id and evt...
-              eventEmitter.emit('eventMissing', id, entry.revision, evt.head ? evt.head.revision : null, evt);
-            }
+  /**
+   * Queues an event with its callback by aggregateId
+   * @param {String}   aggId    The aggregate id.
+   * @param {Object}   evt      The event object.
+   * @param {Function} callback The event callback.
+   */
+  queueEvent: function (aggId, evt, callback) {
+    var self = this;
+    var evtId = dotty.get(evt, this.definition.id);
+    var revInEvt = dotty.get(evt, this.definition.revision);
+    
+    this.queue.push(aggId, evtId, evt, callback, function (loopCount, waitAgain) {
+      self.store.get(aggId, function (err, revInStore) {
+        if (err) {
+          debug(err);
+          self.store.remove(aggId, evtId);
+          return callback(err);
+        }
+
+        if (revInEvt === revInStore) {
+          debug('revision match');
+          callback(null, function (clb) {
+            self.finishGuard(evt, revInStore, clb);
           });
-        });
-      },
-      getQueuedEvents: function(id) {
-        return self.queue.get(id);
-      },
-      removeQueuedEvent: function(id, evt) {
-        self.queue.remove(id, evt);
-      },
-
-      alreadyDenormalized: function(evt, callback) {
-        if (!self.eventQueue) {
-          if (callback) callback(null);
           return;
         }
 
-        self.eventQueue.remove(evt.id, function(err, removed) {
-          if (callback) callback(err);
-        });
-      },
-
-      wouldQueue: function(evt, entry) {
-        if (self.options.ignoreRevision) {
-          return false;
+        if (loopCount < self.options.queueTimeoutMaxLoops) {
+          debug('revision mismatch, try/wait again...');
+          return waitAgain();
         }
+        
+        debug('event timeouted');
+        // try to replay depending from id and evt...
+        var info = {
+          aggregateId: aggId,
+          aggregateRevision: !!self.definition.revision ? dotty.get(evt, self.definition.revision) : undefined,
+          aggregate: !!self.definition.aggregate ? dotty.get(evt, self.definition.aggregate) : undefined,
+          context: !!self.definition.context ? dotty.get(evt, self.definition.context) : undefined,
+          guardRevision: revInStore
+        };
+        self.onEventMissingHandle(info, evt);
+      });
+    });
+  },
 
-        if (!entry.revision) {
-          return false;
+  /**
+   * Finishes the guard stuff and save the new revision to store.
+   * @param {Object}   evt        The event object.
+   * @param {Number}   revInStore The actual revision number in store.
+   * @param {Function} callback   The function, that will be called when this action is completed.
+   *                              `function(err){}`
+   */
+  finishGuard: function (evt, revInStore, callback) {
+    var aggId = dotty.get(evt, this.definition.aggregateId);
+    var evtId = dotty.get(evt, this.definition.id);
+    var revInEvt = dotty.get(evt, this.definition.revision);
+    
+    var self = this;
+    
+    this.store.set(aggId, revInEvt + 1, revInStore, function (err) {
+      if (err) {
+        debug(err);
+        if (err instanceof ConcurrencyError) {
+          var retryIn = randomBetween(0, self.options.retryOnConcurrencyTimeout || 800);
+          debug('retry in ' + retryIn + 'ms');
+          setTimeout(function() {
+            self.guard(evt, callback);
+          }, retryIn);
+          return;
         }
-        if (evt.head && evt.head.revision < entry.revision) {
-          return false;
-        }
-        if (evt.head && evt.head.revision > entry.revision) {
-          return true;
-        }
-        return false;
-      },
-      queuingStrategy: function(evt, entry) {
-        if (self.options.ignoreRevision) {
-          return true;
-        }
 
-        if (!entry.revision) {
-          return true;
-        }
-        if (evt.head && evt.head.revision < entry.revision) {
-          var aux = self._getAux();
-          aux.alreadyDenormalized(evt);
-          return false;
-        }
-        if (evt.head && evt.head.revision > entry.revision) {
-          this.queueEvent(entry.id, evt);
-          return false;
-        }
-        return true;
-      },
-      dequeuingStrategy: function(entry) {
-        var pendingEvents = this.getQueuedEvents(entry.id);
-        if (!pendingEvents) return;
-
-        var nextEvent = _.find(pendingEvents, function(item) {
-          return item.head.revision === entry.revision;
-        });
-        if (!nextEvent) return;
-
-        this.removeQueuedEvent(entry.id, nextEvent); // dequeue event
-        self.guard(nextEvent); // guard event
-      },
-      revisionUpdateStrategy: function(entry, evt) {
-        entry.revision = evt.head.revision + 1;
+        return callback(err);
       }
-    };
 
-    return this._aux;
-  },
+      self.queue.remove(aggId, evtId);
+      callback(null);
+      
+      var pendingEvents = self.queue.get(aggId);
+      if (!pendingEvents || pendingEvents.length === 0) return debug('no other pending event found');
 
-  configure: function(fn) {
-    fn.call(this);
-    return this;
-  },
+      var nextEvent = _.find(pendingEvents, function (e) {
+        var revInEvt = dotty.get(e.payload, self.definition.revision);
+        return revInEvt === revInStore;
+      });
+      if (!nextEvent) return debug('no next pending event found');
 
-  use: function(module) {
-    if (!module) return;
-
-    if (module.getRevision && module.saveRevision) {
-      this.store = module;
-    }
-
-    if (module.dispatch) {
-      this.eventDispatcher = module;
-    }
-
-    if (module.push && module.remove) {
-      this.eventQueue = module;
-    }
+      debug('found next pending event, guard');
+      self.guard(nextEvent.payload, nextEvent.callback);
+    });
   },
 
-  initialize: function(options, callback) {
-    var self = this;
-
-    if (!callback) {
-      callback = options;
-      options = {};
+  /**
+   * Guard this event. Check for order and check if missing events...
+   * @param {Object} evt The event object.
+   * @param {Function} callback The event callback.
+   */
+  guard: function (evt, callback) {
+    if (!this.definition.aggregateId || !dotty.exists(evt, this.definition.aggregateId) ||
+        !this.definition.revision || !dotty.exists(evt, this.definition.revision)) {
+      var err = new Error('Please define an aggregateId!');
+      debug(err);
+      return callback(err);
     }
-
-    var defaults = {
-      ignoreRevision: false,
-      queueTimeout: 3000,
-      queueTimeoutMaxLoops: 3,
-      guardTimeoutMaxLoops: 3
-    };
-
-    _.defaults(options, defaults);
-
-    this.options = options;
-
-    this.queue = new Queue({ queueTimeout: options.queueTimeout });
-
-    if (callback) callback(null);
-  },
-
-  guard: function(evt, callback) {
-    var aux = this._getAux();
-
+    
     var self = this;
+    
+    var aggId = dotty.get(evt, this.definition.aggregateId);
+    var revInEvt = dotty.get(evt, this.definition.revision);
 
-    if (this.options.ignoreRevision || !evt.payload || !evt.payload.id || !evt.head || !evt.head.revision) {
-      this.eventDispatcher.dispatch(evt, callback);
-      return;
-    }
-
-    function proceed(entry) {
-      if (!aux.queuingStrategy(evt, entry)) {
-        if (callback) callback(null);
+    function proceed (revInStore) {
+      if (!revInStore) {
+        debug('first revision to store');
+        callback(null, function (clb) {
+          self.finishGuard(evt, revInStore, clb);
+        });
         return;
       }
+      
+      if (revInEvt < revInStore) {
+        debug('event already denormalized');
+        callback(new AlreadyDenormalizedError(), function (clb) {
+          clb(null);
+        });
+        return;
+      }
+      
+      if (revInEvt > revInStore) {
+        debug('queue event');
+        self.queueEvent(aggId, evt, callback);
+        return;
+      }
 
-      aux.finishGuard(evt, entry, callback);
+      callback(null, function (clb) {
+        self.finishGuard(evt, revInStore, clb);
+      });
     }
 
-    function retry(max, loop) {
-      setTimeout(function() {
-        self.store.getRevision(evt.payload.id, function(err, entry) {
+    function retry (max, loop) {
+      setTimeout(function () {
+        self.store.get(aggId, function(err, revInStore) {
+          if (err) {
+            debug(err);
+            return callback(err);
+          }
+          
           if (loop <= 0) {
-            proceed(entry);
-          } else {
+            return proceed(revInStore);
+          }
+
+          if (!revInStore && revInEvt !== 1) {
             retry(max, --loop);
+            return;
           }
+
+          proceed(revInStore);
         });
       }, randomBetween(max / 5, max));
     }
+    
+    process.nextTick(function () {
+      self.store.get(aggId, function (err, revInStore) {
+        if (err) {
+          debug(err);
+          return callback(err);
+        }
 
-    this.store.getRevision(evt.payload.id, function(err, entry) {
-      if (!entry.revision && evt.head.revision !== 1) {
-        var max = (self.options.queueTimeout * self.options.queueTimeoutMaxLoops) / 3;
-        if (max < 10) {
-          max = 10;
+        if (!revInStore && revInEvt !== 1) {
+          var max = (self.options.queueTimeout * self.options.queueTimeoutMaxLoops) / 3;
+          max = max < 10 ? 10 : max;
+          retry(max, self.options.queueTimeoutMaxLoops);
+          return;
         }
-        retry(max, self.options.guardTimeoutMaxLoops);
-      } else {
-        proceed(entry);
-      }
+
+        proceed(revInStore);
+      });
     });
   }
 
-};
+};
+
+module.exports = RevisionGuard;
+

package.json

 {
   "author": "adrai",
   "name": "cqrs-eventdenormalizer",
-  "version": "0.4.1",
+  "version": "1.0.0",
   "private": false,
@@ -14,13 +14,18 @@ "main": "index.js",
   "dependencies": {
-    "async": ">= 0.9.0",
-    "lodash": ">= 2.4.1",
-    "eventemitter2": ">= 0.4.13",
-    "node-queue": "1.0.x",
-    "viewmodel": "< 1.0.0",
-    "node-uuid": ">= 1.4.1"
+    "async": "0.9.0",
+    "debug": "2.0.0",
+    "dotty": "0.0.2",
+    "viewmodel": "1.1.5",
+    "jsondate": "0.0.1",
+    "lodash": "2.4.1",
+    "node-uuid": "1.4.1",
+    "tolerance": "1.0.0"
   },
   "devDependencies": {
-    "sinon": ">= 1.3.4",
+    "cradle": ">=0.6.7",
+    "expect.js": ">= 0.1.2",
     "mocha": ">= 1.0.1",
-    "expect.js": ">= 0.1.2"
+    "mongodb": ">= 0.0.1",
+    "redis": ">= 0.10.1",
+    "tingodb": ">= 0.0.1"
   },
@@ -52,5 +57,5 @@ "description": "Node-cqrs-eventdenormalizer is a node.js module that implements the cqrs pattern. It can be very useful as eventdenormalizer component if you work with (d)ddd, cqrs, domain, host, etc.",
   ],
-  "scripts" : { 
-    "test" : "mocha && mocha test/integration/eventDenormalizerTest.js"
+  "scripts" : {
+    "test": "mocha test/unit && mocha test/integration"
   } 
 }

README.md

@@ -10,125 +10,496 @@ # Introduction
 
-    $ npm install cqrs-eventdenormalizer
+    npm install cqrs-eventdenormalizer
 
 # Usage
 
-## Initialization
+	var denormalizer = require('cqrs-eventdenormalizer')({
+	  // the path to the "working directory"
+	  // can be structured like
+	  // [set 1](https://github.com/adrai/node-cqrs-eventdenormalizer/tree/master/test/integration/fixture/set1) or
+	  // [set 2](https://github.com/adrai/node-cqrs-eventdenormalizer/tree/master/test/integration/fixture/set2)
+	  denormalizerPath: '/path/to/my/files',
+	  
+	  // optional, default is 'commandRejected'
+	  // will be used to catch AggregateDestroyedError from cqrs-domain
+	  commandRejectedEventName: 'rejectedCommand',
+	  
+	  // optional, default is 800
+	  // if using in scaled systems, this module tries to catch the concurrency issues and
+	  // retries to handle the event after a timeout between 0 and the defined value
+	  retryOnConcurrencyTimeout: 1000,
+	  
+	  // optional, default is in-memory
+	  // currently supports: mongodb, redis, tingodb, couchdb and inmemory
+	  // hint: [viewmodel](https://github.com/adrai/node-viewmodel#connecting-to-any-repository-mongodb-in-the-example--modewrite)
+	  // hint settings like: [eventstore](https://github.com/adrai/node-eventstore#provide-implementation-for-storage)
+	  repository: {
+	    type: 'mongodb',
+	    host: 'localhost',                          // optional
+	    port: 27017,                                // optional
+	    dbName: 'readmodel',                        // optional
+	    timeout: 10000                              // optional
+	    // username: 'technicalDbUser',                // optional
+	    // password: 'secret'                          // optional
+	  },
+     	  
+	  // optional, default is in-memory
+	  // currently supports: mongodb, redis, tingodb and inmemory
+	  // hint settings like: [eventstore](https://github.com/adrai/node-eventstore#provide-implementation-for-storage)
+	  revisionGuard: {
+	    queueTimeout: 1000,                         // optional, timeout for non-handled events in the internal in-memory queue
+	    queueTimeoutMaxLoops: 3                     // optional, maximal loop count for non-handled event in the internal in-memory queue
+	    
+	    type: 'redis',
+	    host: 'localhost',                          // optional
+	    port: 6379,                                 // optional
+	    db: 0,                                      // optional
+	    prefix: 'readmodel_revision',               // optional
+	    timeout: 10000                              // optional
+	    // password: 'secret'                          // optional
+	  }
+	});
 
-	var eventDenormalizer = require('cqrs-eventdenormalizer').eventDenormalizer;
 
-	eventDenormalizer.on('event', function(evt) {
-        // send to clients
-    });
-    eventDenormalizer.initialize({
-        viewBuildersPath: __dirname + '/viewBuilders',
-        extendersPath: __dirname + '/eventExtenders',
-        ignoreRevision: false,
-        disableQueuing: false
-    }, function(err) {
+## Catch connect ad disconnect events
 
-    });
+	// repository
+	denormalizer.repository.on('connect', function() {
+	  console.log('repository connected');
+	});
+	
+	denormalizer.repository.on('disconnect', function() {
+	  console.log('repository disconnected');
+	});
+	
+	// revisionGuardStore
+	denormalizer.revisionGuardStore.on('connect', function() {
+	  console.log('revisionGuardStore connected');
+	});
+	
+	denormalizer.revisionGuardStore.on('disconnect', function() {
+	  console.log('revisionGuardStore disconnected');
+	});
+	
+	
+	// anything (repository or revisionGuardStore)
+	denormalizer.on('connect', function() {
+	  console.log('something connected');
+	});
+	
+	denormalizer.on('disconnect', function() {
+	  console.log('something disconnected');
+	});
 
-    eventDenormalizer.denormalize({ id: 'msgId', event: 'dummyChanged', payload: { id: '23445' } }, function(err) {
 
-    });
+## Define the event structure
+The values describes the path to that property in the event message.
 
-    // if revision checks are enabled
-    eventDenormalizer.on('eventMissing', function(id, aggregateRevision, eventRevision, evt) {
-        // request the appropriate missing events from domain...
-    });
+	denormalizer.defineEvent({
+	  // optional, default is 'correlationId'
+	  // will use the command id as correlationId, so you can match it in the sender
+	  // will be used to copy the correlationId to the notification
+	  correlationId: 'correlationId',
+	  
+	  // optional, default is 'id'
+	  id: 'id',
+	  
+	  // optional, default is 'name'
+	  name: 'name',
+	  
+	  // optional, default is 'aggregate.id'
+	  aggregateId: 'aggregate.id',
+	  
+	  // optional
+	  context: 'context.name',
+	  
+	  // optional
+	  aggregate: 'aggregate.name',
+	  
+	  // optional, default is 'payload'
+	  payload: 'payload',
+	  
+	  // optional, default is 'revision'
+	  // will represent the aggregate revision, can be used in next command
+	  revision: 'revision',
+	  
+	  // optional
+	  version: 'version',
+	  
+	  // optional, if defined the values of the command will be copied to the event (can be used to transport information like userId, etc..)
+	  meta: 'meta'
+	});
 
-    // to replay
-    eventDenormalizer.replay([] /* array of ordered events */, function(err) {});
 
-    // to replay streamed
-    eventDenormalizer.replayStreamed(function(replay, done) {
+## Define the notification structure
+The values describes the path to that property in the notification message.
 
-        replay(evt1);
-        replay(evt2);
-        replay(evt3);
+	denormalizer.defineNotification({
+	  // optional, default is 'correlationId'
+	  // will use the command id as correlationId, so you can match it in the sender
+	  // will be used to copy the correlationId from the event
+	  correlationId: 'correlationId',
+	  
+	  // optional, default is 'id'
+	  id: 'id',
+	  
+	  // optional, default is 'name'
+	  action: 'name',
+	  
+	  // optional, default is 'collection'
+	  collection: 'collection',
+	  
+	  // optional, default is 'payload'
+	  payload: 'payload',
+	  
+	  // optional, will be copied from event
+	  aggregateId: 'meta.aggregate.id',
+	  
+	  // optional, will be copied from event
+	  context: 'meta.context.name',
+	  
+	  // optional, will be copied from event
+	  aggregate: 'meta.aggregate.name',
+	  
+	  // optional, will be copied from event
+	  // will represent the aggregate revision, can be used in next command
+	  revision: 'meta.aggregate.revision',
+	  
+	  // optional, will be copied from event
+	  eventId: 'meta.event.id',
+	  
+	  // optional, will be copied from event
+	  event: 'meta.event.name',
+	  
+	  // optional, if defined the values of the event will be copied to the notification (can be used to transport information like userId, etc..)
+	  meta: 'meta'
+	});
 
-        done(function(err) { });
 
-    });
+## Define the id generator function [optional]
+### you can define a synchronous function
 
-## Define ViewBuilders...
+	denormalizer.idGenerator(function () {
+	  var id = require('node-uuid').v4().toString();
+	  return id;
+	});
 
-    var base = require('cqrs-eventdenormalizer').viewBuilderBase;
+### or you can define an asynchronous function
 
-    module.exports = base.extend({
+	denormalizer.idGenerator(function (callback) {
+	  setTimeout(function () {
+	    var id = require('node-uuid').v4().toString();
+	    callback(null, id);
+	  }, 50);
+	});
 
-        events: [
-            'dummied',
-            {
-                event: 'dummyCreated',
-                method: 'create',
-                viewModelId: 'payload.id'
-            },
-            {
-                event: 'dummyChanged',
-                method: 'update',
-                payload: 'payload'
-            },
-            {
-                event: 'dummyDeleted',
-                method: 'delete'
-            },
-            'dummySpezi',
-            'somethingFlushed',
-            {
-                event: 'versioned'
-            },
-            {
-                event: 'versioned'
-                version: 1
-            }
-        ],
 
-        collectionName: 'dummies',
+## Wire up events [optional]
+### you can define a synchronous function
 
-        dummied: function(data, vm, evt) {
-        },
-  
-        dummySpezi: function(data, vm, evt) {
-          vm.otherValue = 'value';
-        },
-  
-        somethingFlushed: function(data, vm, evt) {
-        },
+	// pass events to bus
+	denormalizer.onEvent(function (evt) {
+	  bus.emit('event', evt);
+	});
+	
+### or you can define an asynchronous function
 
-        versioned: function(data, vm, evt) {
-        },
+	// pass events to bus
+	denormalizer.onEvent(function (evt, callback) {
+	  bus.emit('event', evt, function ack () {
+	    callback();
+	  });
+	});
 
-        versioned_1: function(data, vm, evt) {
-        }
 
-    });
+## Wire up notifications [optional]
+### you can define a synchronous function
 
-## Settings for a scalable solution
+	// pass notifications to bus
+	denormalizer.onNotification(function (noti) {
+	  bus.emit('event', evt);
+	});
+	
+### or you can define an asynchronous function
 
-    {
-        viewBuildersPath: __dirname + '/viewBuilders',
-        extendersPath: __dirname + '/eventExtenders',
-        eventQueue: { type: 'inMemory', collectionName: 'events' },
-        repository: {
-            type: 'mongoDb',
-            dbName: 'mydb',
-            timeout: 60 * 1000
-        },
-        revisionGuardStore: {
-            type: 'mongoDb',
-            dbName: 'mydb',
-            collectionName: 'revisionguard',
-            timeout: 60 * 1000//,
-            //revisionStart: 1
-        },
-        ignoreRevision: false,
-        disableQueuing: false,
-        revisionGuardQueueTimeout: 3000,
-        revisionGuardQueueTimeoutMaxLoops: 3
-    }
+	// pass notifications to bus
+	denormalizer.onNotification(function (noti, callback) {
+	  bus.emit('notification', noti, function ack () {
+	    callback();
+	  });
+	});
 
-See [tests](https://github.com/adrai/node-cqrs-eventdenormalizer/tree/master/test) for detailed information...
 
+## Wire up event missing [optional]
+### you can define a synchronous function
 
+	denormalizer.onEventMissing(function (info, evt) {
+	  console.log(info);
+	  console.log(evt);
+	});
+
+
+## Define default event extension [optional]
+### you can define a synchronous function
+
+	denormalizer.defaultEventExtension(function (evt) {
+	  evt.receiver = [evt.meta.userId];
+	  return evt;
+	});
+	
+### or you can define an asynchronous function
+
+	denormalizer.defaultEventExtension(function (evt, callback) {
+	  evt.receiver = [evt.meta.userId];
+	  callback(null, evt);
+	});
+
+
+## Initialization
+	
+	denormalizer.init(function (err) {
+	  // this callback is called when all is ready...
+	});
+	
+	// or
+	
+	denormalizer.init(); // callback is optional
+
+
+## Handling an event
+
+	denormalizer.handle({
+	  id: 'b80ade36-dd05-4340-8a8b-846eea6e286f',
+	  correlationId: 'c80ada33-dd05-4340-8a8b-846eea6e151d',
+	  name: 'enteredNewPerson',
+	  aggregate: {
+	    id: '3b4d44b0-34fb-4ceb-b212-68fe7a7c2f70',
+	    name: 'person'
+	  },
+	  context: {
+	    name: 'hr'
+	  },
+	  payload: {
+	    firstname: 'Jack',
+	    lastname: 'Huston'
+	  },
+	  revision: 1,
+	  version: 0,
+	  meta: {
+	    userId: 'ccd65819-4da4-4df9-9f24-5b10bf89ef89'
+	  }
+	}); // callback is optional
+	
+### or
+	
+	denormalizer.handle({
+	  id: 'b80ade36-dd05-4340-8a8b-846eea6e286f',
+	  correlationId: 'c80ada33-dd05-4340-8a8b-846eea6e151d',
+	  name: 'enteredNewPerson',
+	  aggregate: {
+	    id: '3b4d44b0-34fb-4ceb-b212-68fe7a7c2f70',
+	    name: 'person'
+	  },
+	  context: {
+	    name: 'hr'
+	  },
+	  payload: {
+	    firstname: 'Jack',
+	    lastname: 'Huston'
+	  },
+	  revision: 1,
+	  version: 0,
+	  meta: {
+	    userId: 'ccd65819-4da4-4df9-9f24-5b10bf89ef89'
+	  }
+	}, function (errs, evt, notifications) {
+	  // this callback is called when event is handled successfully or unsuccessfully
+	  // errs can be of type:
+	  // - null
+	  // - Array of Errors
+	  //
+	  // evt: same as passed in 'onEvent' function
+	  //
+	  // notifications: Array of viewmodel changes
+	});
+
+
+# Components definition
+
+## Collection
+
+	module.exports = require('cqrs-eventdenormalizer').defineCollection({
+	  // optional, default is folder name
+    name: 'personDetail'
+	  
+	  // optional, default ''
+	  defaultPayload: 'payload',
+	  
+	  // indexes: [ // for mongodb
+	  //   'profileId',
+	  //   // or:
+	  //   { profileId: 1 },
+	  //   // or:
+	  //   { index: {profileId: 1}, options: {} }
+	  // ]
+	},
+	
+	// optionally, define some initialization data for new view models...
+	{
+	  emails: ['[e-mail]'],
+	  phoneNumbers: []
+	});
+
+## ViewBuilder
+
+	module.exports = require('cqrs-eventdenormalizer').defineViewBuilder({
+	  // optional, default is file name without extension,
+	  // if name is '' it will handle all events that matches
+	  name: 'enteredNewPerson',
+	  
+	  // optional
+	  aggregate: 'person',
+	  
+	  // optional
+	  context: 'hr',
+	  
+	  // optional, default is 0
+	  version: 2,
+	  
+	  // optional, if not defined or not found it will generate a new viewmodel with new id
+	  id: 'aggregate.id',
+	  
+	  // optional, if not defined it will pass the whole event...
+	  payload: 'payload'
+	}, function (data, vm) { // instead of function you can define a string with default handling ('create', 'update', 'delete')
+	  vm.set('firstname', data.firstname);
+	  vm.set('lastname', data.lastname);
+	});
+
+## EventExtender
+
+### for a collection (in a collection folder)
+
+	module.exports = require('cqrs-eventdenormalizer').defineEventExtender({
+	  // optional, default is file name without extension,
+	  // if name is '' it will handle all events that matches
+	  name: 'enteredNewPerson',
+	  
+	  // optional
+	  aggregate: 'person',
+	  
+	  // optional
+	  context: 'hr',
+	  
+	  // optional, default is 0
+	  // if set to -1, it will ignore the version 
+	  version: 2
+	}, function (evt, col, callback) {
+	  // col.loadViewModel()... or from somewhere else... (col.findViewModels( /* see https://github.com/adrai/node-viewmodel#find */ ))
+	  evt.extended = true;
+	  callback(null, evt);
+	});
+	
+	// or 
+	
+	module.exports = require('cqrs-eventdenormalizer').defineEventExtender({
+	  // optional, default is file name without extension,
+	  // if name is '' it will handle all events that matches
+	  name: 'enteredNewPerson',
+	  
+	  // optional
+	  aggregate: 'person',
+	  
+	  // optional
+	  context: 'hr',
+	  
+	  // optional, default is 0
+	  // if set to -1, it will ignore the version 
+	  version: 2,
+	  
+	  // if defined it will load the viewmodel
+	  id: 'payload.id'
+	},
+	function (evt, vm) {
+	  evt.extended = vm.get('myValue');
+	  return evt;
+	});
+	
+### not for a collection
+
+	module.exports = require('cqrs-eventdenormalizer').defineEventExtender({
+	  // optional, default is file name without extension,
+	  // if name is '' it will handle all events that matches
+	  name: 'enteredNewPerson',
+	  
+	  // optional
+	  aggregate: 'person',
+	  
+	  // optional
+	  context: 'hr',
+	  
+	  // optional, default is 0
+	  // if set to -1, it will ignore the version 
+	  version: 2
+	}, function (evt) {
+	  evt.extended = true;
+	  return evt;
+	});
+	
+	// or
+	
+	module.exports = require('cqrs-eventdenormalizer').defineEventExtender({
+	  // optional, default is file name without extension,
+	  // if name is '' it will handle all events that matches
+	  name: 'enteredNewPerson',
+	  
+	  // optional
+	  aggregate: 'person',
+	  
+	  // optional
+	  context: 'hr',
+	  
+	  // optional, default is 0
+	  // if set to -1, it will ignore the version 
+	  version: 2
+	}, function (evt, callback) {
+	  evt.extended = true;
+	  callback(null, evt);
+	});
+
+
+## Replay events
+
+Replay whenever you want...
+
+	denormalizer.replay([/* ordered array of events */], function (err) {
+	  if (err) { console.log(err); }
+	});
+	
+or when catching some events:
+
+	denormalizer.onEventMissing(function (info, evt) {
+	  
+	  // grab the missing events, depending from info values...
+	  // and call replay...
+	  denormalizer.replay([/* ordered array of events */], function (err) {
+	    if (err) { console.log(err); }
+	  });
+	  
+	});
+
+### streamed
+
+	denormalizer.replayStreamed(function (replay, done) {
+	
+	  replay(evt1);
+	  replay(evt2);
+	  replay(evt3);
+	
+	  done(function (err) {
+	    if (err) { console.log(err); }
+	  });
+	
+	});
+
+
 [Release notes](https://github.com/adrai/node-cqrs-eventdenormalizer/blob/master/releasenotes.md)
@@ -135,0 +506,0 @@

releasenotes.md

@@ -0,35 +1,36 @@
+## v1.0.0
+- refactored whole module
+- added possibility to define aggregateId, aggregate and context
+- generic message structure for events
+- added a lot of tests
+- stabilized everything
+- optimized performance
+- added notificatio feature
+- IMPORTANT: changed API!!!
+
 ## v0.4.1
-
 - do not use newer viewmodel version
 
 ## v0.4.0
-
 - updated node-queue
 
 ## v0.3.6
-
 - make use of viewmodel indexes
 
 ## v0.3.5
-
 - emit missingEvent if commandRejected and revision not in sync
 
 ## v0.3.4
-
 - handle versioned events
 
 ## v0.3.3
-
 - little fix in replay streamed
 
 ## v0.3.2
-
 - introduced optional revisionStart (default = 1)
 
 ## v0.3.1
-
 - optimized guard for first event for a new denormalized aggregate id
 
 ## v0.3.0 (BREAKING CHANGES!!!)
-
 - introduction of revisionGuard
@@ -43,7 +44,5 @@ - contextEventDenormalizer is now eventDenormalizer
 ## v0.2.6
-
 - use new concurrency feature of viewmodel
 
 ## v0.2.4
-
 - added disableQueuing and ignoreRevision flag

New alerts

Major refactor

Supply chain risk

Package has recently undergone a major refactor. It may be unstable or indicate significant internal changes. Use caution when updating to versions that include significant changes.

Found 1 instance in 1 package

Dynamic require

Supply chain risk

Dynamic require can indicate the package is performing dangerous or unsafe dynamic code execution.

Found 1 instance in 1 package

Fixed alerts

Dynamic require

Supply chain risk

Dynamic require can indicate the package is performing dangerous or unsafe dynamic code execution.

Found 1 instance in 1 package

No v1

Quality

Package is not semver >=1. This means it is not stable and does not support ^ ranges.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

169531

increased by145.33%

Number of package files

38

increased by22.58%

Lines of code

2930

increased by147.47%

Number of low quality alerts

0

decreased by-100%

Number of lines in readme file

528

increased by236.31%

Worsened metrics

Dependency count

8

increased by33.33%

Dev dependency count

6

increased by100%

Number of low supply chain risk alerts

6

increased by20%

Number of medium supply chain risk alerts

1

increased byInfinity%

Dependency changes

• + Addeddebug@2.0.0

• + Addeddotty@0.0.2

• + Addedjsondate@0.0.1

• + Addedtolerance@1.0.0

• + Addeddebug@2.0.0(transitive)

• + Addeddotty@0.0.2(transitive)

• + Addedms@0.6.2(transitive)

• + Addedviewmodel@1.1.5(transitive)

• - Removedeventemitter2@>= 0.4.13

• - Removednode-queue@1.0.x

• - Removedasync@3.2.5(transitive)

• - Removedeventemitter2@6.4.9(transitive)

• - Removedlodash@4.17.21(transitive)

• - Removednode-queue@1.0.5(transitive)

• - Removednode-uuid@1.4.8(transitive)

• - Removedviewmodel@0.5.5(transitive)

• Updatedasync@0.9.0

• Updatedlodash@2.4.1

• Updatednode-uuid@1.4.1

• Updatedviewmodel@1.1.5