Alpha

lib/assembler.js

@@ -100,7 +100,7 @@ Assembler.prototype.assembleApp = function (callback) {
     }
   }
 
-  var createApp = function createApp() {
+  var createApp = function createApp(callback) {
     self.app = new self.App(appOptions);
     callback();
   };
@@ -111,10 +111,13 @@ Assembler.prototype.assembleApp = function (callback) {
         return callback(err);
       appOptions.domainRepository = self.domainRepository;
       appOptions.commandBus       = self.commandBus;
-      createApp();
+      createApp(function(){
+        DomainObject.initializeConstructors();
+        callback();
+      });
     });
   } else {
-    createApp();
+    createApp(callback);
   }
 };
 
@@ -241,7 +244,6 @@ Assembler.prototype.tearDownApp = function (callback) {
   var self = this;
 
   // TODO: review this, anything else to unload/destroy?
-
   async.series([
     function (next) {
       self.destroyEventBusEmitter(next);
@@ -277,19 +279,19 @@ Assembler.prototype.destroyReporters = function (callback) {
   if (self.reporters.length == 0)
     return callback();
 
-  var queue = async.queue(function (reporter, callback) {
+  var queue = async.queue(function (reporter, taskCallback) {
     reporter.destroy(function (err) {
-      if (err)
+      if (err) {
         self.logger.error("Assembler#unloadReporters", "destroying reporter failed: " + err);
-      callback(err);
+        return callback(err);
+      }
+      taskCallback();
     });
   }, Infinity);
 
-  queue.drain = function (err) {
-    if (err)
-      self.logger.error("Assembler#unloadReporters", "error: " + err);
+  queue.drain = function() {
     self.reporters = [];
-    callback(err);
+    callback();
   };
   queue.push(self.reporters);
 };
@@ -300,23 +302,23 @@ Assembler.prototype.destroyReportStores = function (callback) {
   if (self.reportStores.length == 0)
     return callback();
 
-  var queue = async.queue(function (reportStore, callback) {
+  var queue = async.queue(function (reportStore, taskCallback) {
     if (reportStore.destroy) {
       reportStore.destroy(function (err) {
-        if (err)
+        if (err) {
           self.logger.error("Assembler#destroyReportStores", "destroying report store failed: " + err);
-        callback(err);
+          return callback(err);
+        }
+        taskCallback();
       });
     } else {
-      callback();
+      taskCallback();
     }
   }, Infinity);
 
-  queue.drain = function (err) {
-    if (err)
-      self.logger.error("Assembler#destroyReportStores", "error: " + err);
+  queue.drain = function() {
     self.reportStores = [];
-    callback(err);
+    callback();
   };
   queue.push(self.reportStores);
 };
@@ -411,8 +413,6 @@ Assembler.prototype._assembleDomainInfrastructure = function (callback) {
       logger:           self.logger
     });
 
-    DomainObject.initializeConstructors();
-
     self.eventBusEmitter  = eventBusEmitter;
     self.domainRepository = domainRepository;
     self.commandBus       = commandBus;

lib/command_bus.coffee

@@ -57,6 +57,7 @@ class CommandBus
                 p2.end()
                 p.end()
                 done args...
+          transaction.callback = callback
         else
           transaction = (done) ->
             p.start()

lib/command_bus_client.coffee

@@ -36,8 +36,7 @@ class CommandBusClient
     request = @_makeRequest path: "/commands"
     stream  = request.stream
 
-    request.on "error", (err) ->
-      callback err
+    request.on "error", callback
 
     request.on "response", (response) =>
       @_processResponse response, callback
@@ -96,4 +95,4 @@ class CommandBusClient
         logger.error "CommandBusClient", "command failed remotely: #{error.stack || error.message || error}"
         callback error
 
-module.exports = CommandBusClient
+module.exports = CommandBusClient

lib/command_bus_server.coffee

@@ -87,7 +87,6 @@ class CommandBusServer
 
       logger.log "CommandBusServer", "deserialize command \"#{commandName}\""
       @commandBus.deserializeCommand commandName, payload, (err, command) =>
-        logger.log "CommandBusServer", "start command \"#{commandName}\""
         @commandBus.executeCommand command, (err) ->
           if err?
             logger.alert "CommandBusServer", "error while executing command (#{err})"

lib/command_handler.js

@@ -28,6 +28,7 @@ CommandHandler.registerAllInDirectoryOnBus = function (path, options) {
   var logger     = options.logger;
   fs.readdirSync(path).forEach(function (fileName) {
     var CommandHandler = require(path + "/" + fileName);
+    CommandHandler.logger = logger;
     logger.log("CommandHandler", "loading command handler for command \"" + CommandHandler.getCommandName() + "\" (" + fileName + ")");
     commandBus.registerCommandHandler(CommandHandler);
   });

lib/domain_repository.coffee

@@ -22,6 +22,7 @@ class DomainRepository
     @transactionQueue      = async.queue (transaction, done) =>
       if @halted
         @logger.warning "transaction", "skipped (#{@transactionQueue.length()} more transaction(s) in queue)"
+        transaction.callback() if transaction.callback?
         done()
       else
         @transacting = true
@@ -75,16 +76,10 @@ class DomainRepository
     entityInstantiator.findByUid uid, callback
 
   replayAllEvents: (callback) ->
-    @store.findAllEvents (err, events) =>
-      if events.length > 0
-        eventQueue = async.queue (event, eventTaskCallback) =>
-          event.replayed = true
-          @_publishEvent event, eventTaskCallback
-        , 1
-        eventQueue.drain = callback
-        eventQueue.push events
-      else
-        callback()
+    @store.findAllEventsOneByOne (err, event, eventHandlerCallback) =>
+      event.replayed = true
+      @_publishEvent event, eventHandlerCallback
+    , callback
 
   getLastPublishedEvents: () ->
     @emitter.lastEmittedEvents

lib/entity.coffee

@@ -35,9 +35,9 @@ Entity = (name, finalCtor, Ctor) ->
   Base::_initializeAtVersion = (version) ->
     @$version          = version
     @$appliedEvents    = []
-    @$domainRepository = finalCtor.$domainRepository
-    @$commandBus       = finalCtor.$commandBus
-    @$logger           = finalCtor.$logger
+    @$domainRepository = Entity.domainRepository
+    @$commandBus       = Entity.commandBus
+    @$logger           = Entity.logger
 
   Base::_serialize = (contained) ->
     throw new Error "References between entity are forbidden" if contained
@@ -96,12 +96,11 @@ Entity = (name, finalCtor, Ctor) ->
     deferred = Q.defer()
     deferred.promise.nodeify callback
     if uid?
-      @$domainRepository.findEntityByUid @, uid, (err, entity) ->
+      @$domainRepository.findEntityByUid @, uid, (err, entity) =>
         if err?
           deferred.reject err
-        else if not entity
-          deferred.reject new Error "Could not find entity with UID " + uid
         else
+          @$logger.error "findByUid", "Could not find entity with UID #{uid}" if not entity
           deferred.resolve entity
     else
       deferred.reject(new Error 'Please provide a UID to find')

lib/event_bus/common/receiver.js

@@ -69,25 +69,26 @@ CommonEventBusReceiver.prototype._handleEvent = function (event, callback) {
     return callback();
   }
 
-  var parallelHandling = async.queue(function (eventHandler, callback) {
-      eventHandler(event, function (err) {
-        if (err) {
-          self.logger.warning("CommonEventBusReceiver#_handleEvent", "an error occurred in an event handler: " + (err.stack || err.message || err));
-          errors.push(err);
+  var pendingHandlers = 0;
+  eventHandlers.forEach(function(eventHandler){
+    pendingHandlers++
+    eventHandler(event, function (err) {
+      if (err) {
+        self.logger.warning("CommonEventBusReceiver#_handleEvent", "an error occurred in an event handler: " + (err.stack || err.message || err));
+        errors.push(err);
+      }
+      pendingHandlers--;
+      if (pendingHandlers == 0) {
+        if (errors.length > 0) {
+          var error = new Error("Some errors occured when handling the event: [" + errors.join(" | ") + "]");
+          callback(error);
+        } else {
+          self.lastEvent = event;
+          callback(null);
         }
-        callback(err);
-      });
-  }, Infinity);
-
-  parallelHandling.drain = function () {
-    if (errors.length > 0) {
-      var err = new Error("Some errors occured when handling the event: [" + errors.join(" | ") + "]");
-      return callback(err);
-    }
-    self.lastEvent = event;
-    callback();
-  };
-  parallelHandling.push(eventHandlers);
+      }
+    });
+  });
 };
 
 CommonEventBusReceiver.prototype.stop = function (callback) {

lib/event_store/base.coffee

@@ -9,6 +9,9 @@ class BaseEventStore
   findAllEvents: (options, callback) ->
     throw new Error "implement me"
 
+  findAllEventsOneByOne: (options, callback) ->
+    throw new Error "implement me"
+
   findAllEventsByEntityUid: (entityUid, options, callback) ->
     throw new Error "Implement me"
 
@@ -23,4 +26,4 @@ class BaseEventStore
     # implement me if you want snapshots in your store
     callback? null
 
-module.exports = BaseEventStore
+module.exports = BaseEventStore

lib/event_store/mongodb.coffee

@@ -43,14 +43,19 @@ class MongoDbEventStore extends Base
 
   destroy: (callback) ->
     @_closeConnectionAndReturn @db, null, (err) =>
+      return callback err if err?
       @db                 = null
       @eventCollection    = null
       @snapshotCollection = null
       callback null
 
   _closeConnectionAndReturn: (db, err, callback) ->
-    db.close() if db?
-    callback err
+    if db?
+      db.close (closeErr) ->
+        return callback closeErr if closeErr?
+        callback err
+    else
+      callback err
 
   setup: (callback) ->
     async.series [
@@ -70,18 +75,22 @@ class MongoDbEventStore extends Base
     uid = uuid.v4()
     callback null, uid
 
-  findAllEvents: (callback) ->
+  findAllEventsOneByOne: (eventHandler, callback) ->
     p = new Profiler "MongoDbEventStore#_find(db request)", @logger
     p.start()
-    @eventCollection.find({}).sort("timestamp":1).toArray (err, items) =>
-      p.end()
-
-      if err?
-        callback err
-      else if not items?
-        callback null, []
-      else
-        @_instantiateEventsFromRows items, callback
+    cursor = @eventCollection.find({}).sort("timestamp":1)
+    retrieve = =>
+      cursor.nextObject (err, item) =>
+        return callback err if err?
+        if item?
+          @_instantiateEventFromRow item, (err, event) ->
+            eventHandler err, event, (err) ->
+              return callback err if err?
+              retrieve()
+        else
+          p.end()
+          callback null
+    retrieve()
 
   findAllEventsByEntityUid: (entityUid, order, callback) ->
     [order, callback] = [null, order] unless callback?
@@ -211,27 +220,8 @@ class MongoDbEventStore extends Base
     return callback null, events if rows.length is 0
 
     rowsQueue = async.queue (row, rowCallback) =>
-      uid          = row.uid
-      name         = row.name
-      entityUid = row.entityUid
-      data         = row.data
-      timestamp    = row.timestamp
-      version      = row.version
-
-      @_loadAttachmentsFromRow row, (err, attachments) ->
-        return rowCallback err if err?
-
-        for attachmentName, attachmentBody of attachments
-          data[attachmentName] = attachmentBody
-
-        event = new Event
-          name: name
-          data: data
-          uid: uid
-          entityUid: entityUid
-          timestamp: timestamp
-          version: version
-
+      @_instantiateEventFromRow row, (err, event) ->
+        return callback err if err?
         events.push event
         defer rowCallback
     , 1
@@ -241,6 +231,30 @@ class MongoDbEventStore extends Base
 
     rowsQueue.push rows
 
+  _instantiateEventFromRow: (row, callback) ->
+    uid          = row.uid
+    name         = row.name
+    entityUid = row.entityUid
+    data         = row.data
+    timestamp    = row.timestamp
+    version      = row.version
+
+    @_loadAttachmentsFromRow row, (err, attachments) ->
+      return rowCallback err if err?
+
+      for attachmentName, attachmentBody of attachments
+        data[attachmentName] = attachmentBody
+
+      event = new Event
+        name: name
+        data: data
+        uid: uid
+        entityUid: entityUid
+        timestamp: timestamp
+        version: version
+
+      callback null, event
+
   _loadAttachmentsFromRow: (row, callback) ->
     attachments = {}
     for attachmentName, attachmentBody of row._attachments

lib/main.js

@@ -1,4 +1,4 @@
-require("coffee-script");
+require("coffee-script/register");
 
 module.exports = {
   Assembler: require("./assembler"),

package.json

@@ -18,11 +18,11 @@
     "framework"
   ],
   "dependencies": {
-    "coffee-script": "1.3.3",
+    "coffee-script": "1.7.1",
     "async": "0.1.22",
     "node-uuid": "1.3.3",
     "redis": "0.9.0",
-    "hiredis": "~0.1.14",
+    "hiredis": "0.1.16",
     "nano": "~3.3.6",
     "devnull": "0.0.10",
     "mongodb": "~1.2.12",