mkae position option more generic, and mappabale like other options

Author: nanov

lib/base.js

@@ -53,6 +53,16 @@ _.extend(Store.prototype, {
     if (callback) callback(null, id);
   },
 
+  /**
+   * Use this function to an array containing the next position numbers
+   * @param  {number} positins Number of positions to provide.
+   * @param  {Function} callback The function, that will be called when the this action is completed.
+   *                             `function(err, positions){}` positions is either undefined if option is not enabled/supported or array with positions
+   */
+  getNextPositions: function(positions, callback) {
+    callback(null);
+  },
+
   /**
    * loads the events
    * @param {Object}   query    the query object

lib/databases/inmemory.js

@@ -63,6 +63,18 @@ _.extend(InMemory.prototype, {
     if (callback) callback(null);
   },
 
+  getNextPositions: function(positions, callback) {
+    if (!this.options.trackPosition) {
+      return callback(null);
+    }
+    
+    var range = [];
+    for(var i=0; i<positions; i++) {
+      range.push(++this.position);
+    }
+    callback(null, range);
+  },
+
   addEvents: function (events, callback) {
     if (!events || events.length === 0) {
       callback(null);
@@ -96,9 +108,6 @@ _.extend(InMemory.prototype, {
 
     var self = this;
     _.forEach(events, function(evt) {
-      if (self.options.trackPosition) {
-        evt.position = ++self.position;
-      }
       self.undispatchedEvents._direct[evt.id] = evt;
     });
 

lib/databases/mongodb.js

@@ -314,12 +314,15 @@ _.extend(Mongo.prototype, {
       invalidCommitId = false;
 
     var self = this;
-    this.getNextPositions(events.length, function(err, positions) {
+    // this.getNextPositions(events.length, function(err, positions) {
+      /*
       if (err) {
         debug(err);
         if (callback) callback(err);
         return;
       }
+      */
+
       _.forEach(events, function (evt, index) {
         if (!evt.aggregateId) {
           noAggregateId = true;
@@ -331,8 +334,6 @@ _.extend(Mongo.prototype, {
 
         evt._id = evt.id;
         evt.dispatched = false;
-        if (positions)
-          evt.position = positions[index];
       });
 
       if (noAggregateId) {
@@ -378,7 +379,7 @@ _.extend(Mongo.prototype, {
           self.removeTransactions(events[events.length - 1], callback);
         });
       });
-    });
+    // });
   },
 
   // streaming API

lib/event.js

@@ -44,6 +44,7 @@ function Event (eventstream, event, eventMappings) {
   this.commitSequence = null;
   this.commitStamp = null;
   this.payload = event || null;
+  this.position = null;
 
   this.applyMappings = function applyMappings() {
     _.keys(eventMappings).forEach(function (key) {

lib/eventstore.js

@@ -487,38 +487,46 @@ _.extend(Eventstore.prototype, {
           currentRevision = eventstream.currentRevision(),
           uncommittedEvents = [].concat(eventstream.uncommittedEvents);
         eventstream.uncommittedEvents = [];
-        for (var i = 0, len = uncommittedEvents.length; i < len; i++) {
-          event = uncommittedEvents[i];
-          event.id = id + i.toString();
-          event.commitId = id;
-          event.commitSequence = i;
-          event.restInCommitStream = len - 1 - i;
-          event.commitStamp = new Date();
-          currentRevision++;
-          event.streamRevision = currentRevision;
-
-          event.applyMappings();
-        }
 
-        self.store.addEvents(uncommittedEvents, function(err) {
-          if (err) {
-            // add uncommitted events back to eventstream
-            eventstream.uncommittedEvents = uncommittedEvents.concat(eventstream.uncommittedEvents);
-            return callback(err);
+        self.store.getNextPositions(uncommittedEvents.length, function(err, positions) {
+          if (err)
+            return callback(err)
+
+          for (var i = 0, len = uncommittedEvents.length; i < len; i++) {
+            event = uncommittedEvents[i];
+            event.id = id + i.toString();
+            event.commitId = id;
+            event.commitSequence = i;
+            event.restInCommitStream = len - 1 - i;
+            event.commitStamp = new Date();
+            currentRevision++;
+            event.streamRevision = currentRevision;
+            if (positions)
+              event.position = positions[i];
+
+            event.applyMappings();
           }
 
-          if (self.publisher && self.dispatcher) {
-            // push to undispatchedQueue
-            self.dispatcher.addUndispatchedEvents(uncommittedEvents);
-          } else {
-            eventstream.eventsToDispatch = [].concat(uncommittedEvents);
-          }
+          self.store.addEvents(uncommittedEvents, function(err) {
+            if (err) {
+              // add uncommitted events back to eventstream
+              eventstream.uncommittedEvents = uncommittedEvents.concat(eventstream.uncommittedEvents);
+              return callback(err);
+            }
 
-          // move uncommitted events to events
-          eventstream.events = eventstream.events.concat(uncommittedEvents);
-          eventstream.currentRevision();
+            if (self.publisher && self.dispatcher) {
+              // push to undispatchedQueue
+              self.dispatcher.addUndispatchedEvents(uncommittedEvents);
+            } else {
+              eventstream.eventsToDispatch = [].concat(uncommittedEvents);
+            }
 
-          callback(null, eventstream);
+            // move uncommitted events to events
+            eventstream.events = eventstream.events.concat(uncommittedEvents);
+            eventstream.currentRevision();
+
+            callback(null, eventstream);
+          });
         });
       }],
 

test/eventstoreTest.js

@@ -876,6 +876,7 @@ describe('eventstore', function () {
 
       var types = ['inmemory', 'tingodb', 'mongodb', 'redis'/*, 'elasticsearch', 'azuretable', 'dynamodb'*/];
       var streamingApiTypes = ['mongodb'];
+      var positionTypes = ['mongodb', 'inmemory'];
 
       var token = crypto.randomBytes(16).toString('hex');
 
@@ -1582,7 +1583,62 @@ describe('eventstore', function () {
                 });                    
               }
 
+              if (positionTypes.indexOf(type) !== -1) {
+                describe('setting event position option', function() {
+                  beforeEach(function (done) {
+                    es = eventstore({
+                      type: type,
+                      positionsCollectionName: 'positions',
+                      trackPosition: true,
+                    });
+                    es.defineEventMappings({ commitStamp: 'head.position' });
+                    es.init(function(err) {
+                      es.store.clear(done);
+                    });
+                  });
+    
+                  afterEach(function (done) {
+                    es.store.clear(done);
+                  });
+
+                  it('it should save the event with position', function(done) {
+                    es.getEventStream('streamIdWithPosition', function (err, stream) {
+                      expect(err).not.to.be.ok();
+                      stream.addEvent({ one: 'event' });
+                      stream.addEvent({ one: 'event-other' });
+          
+                      stream.commit(function (err, st) {
+                        expect(err).not.to.be.ok();
+          
+                        expect(st.events.length).to.eql(2);
+                        expect(st.events[0].position).to.eql(1);
+                        expect(st.events[1].position).to.eql(2);
+          
+                        done();
+                      });
+                    });
+                  });
+
+                  it('it should map position to payload', function(done) {
+                    es.getEventStream('streamIdWithPosition', function (err, stream) {
+                      expect(err).not.to.be.ok();
+                      stream.addEvent({ one: 'event' });
+                      stream.addEvent({ one: 'event-other' });
+          
+                      stream.commit(function (err, st) {
+                        expect(err).not.to.be.ok();
+          
+                        expect(st.events.length).to.eql(2);
+                        expect(st.events[0].payload.head.position).to.eql(1);
+                        expect(st.events[1].payload.head.position).to.eql(2);
+          
+                        done();
+                      });
+                    });
+                  });
 
+                });
+              }
             });
 
           });

test/storeTest.js

@@ -3126,101 +3126,6 @@ types.forEach(function (type) {
             })
 
           });
-
-          if (type === "mongodb" || type === "inmemory") {
-            var positionstore = new Store({
-              positionsCollectionName: 'positions',
-              trackPosition: true,
-            });
-    
-            describe('adding position number', function(done) {
-              beforeEach(function (done) {
-                positionstore.connect(done);
-              });
-
-              afterEach(function (done) {
-                positionstore.clear(done);
-              });              
-              describe('with one event in the array', function () {
-
-                it('it should save the event with position', function(done) {
-  
-                  var event = {
-                    aggregateId: 'id1',
-                    id: '111',
-                    streamRevision: 0,
-                    commitId: '111',
-                    commitStamp: new Date(),
-                    commitSequence: 0,
-                    payload: {
-                      event:'bla',
-                      array: []
-                    },
-                    applyMappings: function () {}
-                  };
-  
-                  positionstore.addEvents([event], function(err) {
-                    expect(err).not.to.be.ok();
-  
-                    positionstore.getEvents({}, 0, -1, function(err, evts) {
-                      expect(err).not.to.be.ok();
-                      expect(evts[0].position).to.eql(1);
-                      done();
-                    });
-                  });
-  
-                });
-  
-              });
-  
-              describe('with multiple events in the array', function () {
-  
-                it('it should save the events with positions', function(done) {
-  
-                  var event1 = {
-                    aggregateId: 'id2',
-                    streamRevision: 0,
-                    id: '112',
-                    commitId: '987',
-                    commitStamp: new Date(Date.now() + 1),
-                    commitSequence: 0,
-                    restInCommitStream: 1,
-                    payload: {
-                      event:'bla'
-                    },
-                    applyMappings: function () {}
-                  };
-  
-                  var event2 = {
-                    aggregateId: 'id2',
-                    streamRevision: 1,
-                    id:'113',
-                    commitId: '987',
-                    commitStamp: new Date(Date.now() + 1),
-                    commitSequence: 1,
-                    restInCommitStream: 0,
-                    payload: {
-                      event:'bla2'
-                    },
-                    applyMappings: function () {}
-                  };
-  
-                  positionstore.addEvents([event1, event2], function(err) {
-                    expect(err).not.to.be.ok();
-  
-                    positionstore.getEvents({}, 0, -1, function(err, evts) {
-                      expect(err).not.to.be.ok();
-                      expect(evts).to.be.an('array');
-                      expect(evts).to.have.length(2);
-                      expect(evts[0].position).to.eql(1);
-                      expect(evts[1].position).to.eql(2);
-                      done();
-                    });
-                  });
-                });
-              });
-            });
-          }
         });
       });
     });