lib/eventstore.js
function(eventstream, callback) {

    var self = this;

    async.waterfall([

      function getNewCommitId(callback) {
        self.getNewId(callback);
      },

      function commitEvents(id, callback) {
        // start committing.
        var event,
          currentRevision = eventstream.currentRevision(),
          uncommittedEvents = [].concat(eventstream.uncommittedEvents);
        eventstream.uncommittedEvents = [];

        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();
          }

          self.store.addEvents(uncommittedEvents, function(err) {
            if (err) {
              // add uncommitted events back to eventstream
              eventstream.uncommittedEvents = uncommittedEvents.concat(eventstream.uncommittedEvents);
              return callback(err);
            }

            if (self.publisher && self.dispatcher) {
              // push to undispatchedQueue
              self.dispatcher.addUndispatchedEvents(uncommittedEvents);
            } else {
              eventstream.eventsToDispatch = [].concat(uncommittedEvents);
            }

            // move uncommitted events to events
            eventstream.events = eventstream.events.concat(uncommittedEvents);
            eventstream.currentRevision();

            callback(null, eventstream);
          });
        });
      }],

      callback
    );
  }
function(eventstream, callback) {

    var self = this;

    async.waterfall([

      function getNewCommitId(callback) {
        self.getNewId(callback);
      },

      function commitEvents(id, callback) {
        // start committing.
        var event,
          currentRevision = eventstream.currentRevision(),
          uncommittedEvents = [].concat(eventstream.uncommittedEvents);
        eventstream.uncommittedEvents = [];

        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();
          }

          self.store.addEvents(uncommittedEvents, function(err) {
            if (err) {
              // add uncommitted events back to eventstream
              eventstream.uncommittedEvents = uncommittedEvents.concat(eventstream.uncommittedEvents);
              return callback(err);
            }

            if (self.publisher && self.dispatcher) {
              // push to undispatchedQueue
              self.dispatcher.addUndispatchedEvents(uncommittedEvents);
            } else {
              eventstream.eventsToDispatch = [].concat(uncommittedEvents);
            }

            // move uncommitted events to events
            eventstream.events = eventstream.events.concat(uncommittedEvents);
            eventstream.currentRevision();

            callback(null, eventstream);
          });
        });
      }],

      callback
    );
  }
function ( eventstream , callback ) { var self = this ; async . waterfall ( [ function getNewCommitId ( callback ) { self . getNewId ( callback ) ; } , function commitEvents ( id , callback ) { // start committing. var event , currentRevision = eventstream . currentRevision ( ) , uncommittedEvents = [ ] . concat ( eventstream . uncommittedEvents ) ; eventstream . uncommittedEvents = [ ] ; 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 ( ) ; } self . store . addEvents ( uncommittedEvents , function ( err ) { if ( err ) { // add uncommitted events back to eventstream eventstream . uncommittedEvents = uncommittedEvents . concat ( eventstream . uncommittedEvents ) ; return callback ( err ) ; } if ( self . publisher && self . dispatcher ) { // push to undispatchedQueue self . dispatcher . addUndispatchedEvents ( uncommittedEvents ) ; } else { eventstream . eventsToDispatch = [ ] . concat ( uncommittedEvents ) ; } // move uncommitted events to events eventstream . events = eventstream . events . concat ( uncommittedEvents ) ; eventstream . currentRevision ( ) ; callback ( null , eventstream ) ; } ) ; } ) ; } ] , callback ) ; }
commits all uncommittedEvents in the eventstream
@param eventstream the eventstream that should be saved (hint: directly use the commit function on eventstream)
@param {Function}  callback the function that will be called when this action has finished
`function(err, eventstream){}` (hint: eventstream.eventsToDispatch)
commits all uncommittedEvents in the eventstream