introduced replay stremed

.npmignore

@@ -0,0 +1 @@
+test

README.md

@@ -39,6 +39,17 @@ It can be very useful as eventdenormalizer component if you work with (d)ddd, cq
     // to replay
     eventDenormalizer.replay([] /* array of ordered events */, function(err) {});
 
+    // to replay streamed
+    eventDenormalizer.replayStreamed(function(replay, done) {
+
+        replay(evt1);
+        replay(evt2);
+        replay(evt3);
+
+        done(function(err) { });
+
+    });
+
 ## Define ViewBuilders...
 
     var base = require('cqrs-eventdenormalizer').viewBuilderBase;

lib/bases/viewBuilderBase.js

@@ -173,6 +173,49 @@ ViewBuilder.prototype = {
 
     });
 
+  },
+
+  replayStreamed: function(fn, retryTimout) {
+
+    var self = this;
+
+    retryTimout = retryTimout || 10;
+
+    var queue = [];
+
+    var replay = function(evt) {
+      queue.push(evt);
+
+      (function handle(e) {
+        self.handle(e, true, function(err) {
+          queue.splice(_.indexOf(e), 1);
+
+          if (queue.length > 0) {
+            handle(queue[0]);
+          }
+        });
+      })(evt);
+    };
+
+    var done = function(callback) {
+      (function retry() {
+        if (queue.length > 0) {
+          return setTimeout(retry, retryTimout);
+        }
+
+        var replVms = _.values(self.replayingVms);
+
+        async.each(replVms, function(vm, callback) {
+          self.saveViewModel(vm, callback);
+        }, function(err) {
+          self.replayingVms = {};
+          callback(err);
+        });
+      })();
+    };
+
+    fn(replay, done);
+
   }
 
 };

lib/eventDenormalizer.js

@@ -6,12 +6,13 @@ var viewBuilderLoader = require('./loaders/viewBuilderLoader'),
    _ = require('lodash'),
    async = require('async'),
    queue = require('node-queue'),
-   eventQueue,
-   guardStore,
-   viewBuilders,
+   replayHandler = require('./replayHandler'),
    repository = require('viewmodel').write.create(),
    revisionGuardStore = require('./revisionGuardStore'),
    revisionGuard = require('./revisionGuard'),
+   eventQueue,
+   guardStore,
+   viewBuilders,
    evtDen;
 
 module.exports = evtDen = _.extend(new EventEmitter2({
@@ -103,6 +104,9 @@ module.exports = evtDen = _.extend(new EventEmitter2({
           this.use(eventDispatcher);
           this.use(eventQueue);
         });
+
+        replayHandler.initialize(viewBuilders, guardStore);
+
         revisionGuard.initialize({
           ignoreRevision: options.ignoreRevision,
           queueTimeout: options.revisionGuardQueueTimeout,
@@ -143,51 +147,8 @@ module.exports = evtDen = _.extend(new EventEmitter2({
     }
   },
 
-  replay: function(evts, callback) {
+  replay: replayHandler.replay,
 
-    var revisionMap = {},
-        groupedEvents = {};
-
-    _.each(evts, function(evt) {
-      if (evt.head && evt.head.revision) {
-        revisionMap[evt.payload.id] = evt.head.revision;
-      }
-
-      var interested = _.filter(viewBuilders, function(vB) {
-        return _.contains(vB.registeredEventNames, evt.event);
-      });
-
-      _.each(interested, function(inter) {
-        groupedEvents[inter.id] = groupedEvents[inter.id] || [];
-        groupedEvents[inter.id].push(evt);
-      });
-    });
-
-    async.series([
-      function(callback) {
-        async.each(viewBuilders, function(viewBuilder, callback) {
-          if (!groupedEvents[viewBuilder.id] || groupedEvents[viewBuilder.id].length === 0) {
-            return callback(null);
-          }
-
-          viewBuilder.replay(groupedEvents[viewBuilder.id], 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];
-            guardStore.saveRevision(entry, callback);
-          });
-        }, callback);
-      }
-    ], function(err) {
-      if (callback) callback(err);
-    });
-  }
+  replayStreamed: replayHandler.replayStreamed
 
 });

lib/replayHandler.js

@@ -0,0 +1,119 @@
+var _ = require('lodash'),
+		async = require('async'),
+		viewBuilders,
+		guardStore;
+
+module.exports = {
+	initialize: function(options, vB, gS) {
+		if (!gS) {
+			gS = vB;
+			vB = options;
+			options = {};
+		}
+
+		viewBuilders = vB;
+
+		guardStore = gS;
+	},
+
+	replay: function(evts, callback) {
+
+    var revisionMap = {},
+        groupedEvents = {};
+
+    _.each(evts, function(evt) {
+      if (evt.head && evt.head.revision) {
+        revisionMap[evt.payload.id] = evt.head.revision;
+      }
+
+      var interested = _.filter(viewBuilders, function(vB) {
+        return _.contains(vB.registeredEventNames, evt.event);
+      });
+
+      _.each(interested, function(inter) {
+        groupedEvents[inter.id] = groupedEvents[inter.id] || [];
+        groupedEvents[inter.id].push(evt);
+      });
+    });
+
+    async.series([
+      function(callback) {
+        async.each(viewBuilders, function(viewBuilder, callback) {
+          if (!groupedEvents[viewBuilder.id] || groupedEvents[viewBuilder.id].length === 0) {
+            return callback(null);
+          }
+
+          viewBuilder.replay(groupedEvents[viewBuilder.id], 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];
+            guardStore.saveRevision(entry, callback);
+          });
+        }, callback);
+      }
+    ], function(err) {
+      if (callback) callback(err);
+    });
+  },
+
+  replayStreamed: function(fn, retryTimout) {
+
+    var revisionMap = {},
+        viewBuilderReplayStreams = {};
+
+    var replay = function(evt) {
+      if (evt.head && evt.head.revision) {
+        revisionMap[evt.payload.id] = evt.head.revision;
+      }
+
+      var interested = _.filter(viewBuilders, function(vB) {
+        return _.contains(vB.registeredEventNames, evt.event);
+      });
+
+      _.each(interested, function(inter) {
+        if (!viewBuilderReplayStreams[inter.id]) {
+          inter.replayStreamed(function(vbReplay, vbDone) {
+            viewBuilderReplayStreams[inter.id] = {
+              replay: vbReplay,
+              done: vbDone
+            };
+          }, retryTimout);
+        }
+
+        viewBuilderReplayStreams[inter.id].replay(evt);
+      });
+    };
+
+    var done = function(callback) {
+      async.series([
+        function(callback) {
+          async.each(_.values(viewBuilderReplayStreams), function(viewBuilderReplayStream, callback) {
+            viewBuilderReplayStream.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];
+              guardStore.saveRevision(entry, callback);
+            });
+          }, callback);
+        }
+      ], function(err) {
+        if (callback) callback(err);
+      });
+    };
+
+    fn(replay, done);
+  }
+};

package.json

@@ -1,7 +1,7 @@
 {
   "author": "adrai",
   "name": "cqrs-eventdenormalizer",
-  "version": "0.3.2",
+  "version": "0.3.3",
   "private": false,
   "main": "index.js",
   "engines": {

test/integration/eventDenormalizerTest.js

@@ -778,4 +778,99 @@ describe('EventDenormalizer', function() {
 
     });
 
+    describe('repalying streamed', function() {
+
+        var evts = [
+            {
+                id: '222222-111111',
+                event: 'dummyCreated',
+                head: {
+                    revision: 1
+                },
+                payload: {
+                    id: '222222-1987',
+                    first: 'when created'
+                }
+            },
+            {
+                id: '222222-222222',
+                event: 'loaded',
+                head: {
+                    revision: 1
+                },
+                payload: {
+                    id: '222222-2014'
+                }
+            },
+            {
+                id: '222222-333333',
+                event: 'dummyChanged',
+                head: {
+                    revision: 2
+                },
+                payload: {
+                    id: '222222-1987',
+                    second: 'when updated'
+                }
+            },
+            {
+                id: '222222-444444',
+                event: 'somethingFlushed',
+                head: {
+                    revision: 2
+                },
+                payload: {
+                    id: '222222-2014'
+                }
+            }
+        ];
+
+        it('it should work as expected', function(done) {
+
+            eventDenormalizer.replayStreamed(function(replay, finished) {
+                for (var i = 0, length = evts.length; i < length; i++) {
+                    replay(evts[i]);
+                }
+                finished(function(err) {
+                    async.series([
+                        function(callback) {
+                            dummyRepo.get('222222-1987', function(err, res) {
+                                expect(res.first).to.eql('when created');
+                                expect(res.second).to.eql('when updated');
+
+                                callback(err);
+                            });
+                        },
+                        function(callback) {
+                            dummyRepo.get('222222-2014', function(err, res) {
+                                expect(res.loadedSet).to.eql(undefined);
+                                expect(res.flushSet).to.eql(undefined);
+
+                                callback(err);
+                            });
+                        },
+                        function(callback) {
+                            dummy2Repo.get('222222-2014', function(err, res) {
+                                expect(res.loadedSet).to.eql('loaded');
+                                expect(res.flushSet).to.eql('flushed');
+
+                                callback(err);
+                            });
+                        },
+                        function(callback) {
+                            dummy2Repo.get('222222-1987', function(err, res) {
+                                expect(res.first).to.eql(undefined);
+                                expect(res.second).to.eql(undefined);
+
+                                callback(err);
+                            });
+                        }
+                    ], done);
+                });
+            });
+
+        });
+
+    });
+
 });