Beta

lib/assembler.js

@@ -8,12 +8,15 @@ var AggregateRoot    = require("./aggregate_root");
 var Service          = require("./service");
 var logger           = require("./logger");
 
-function Assembler(App, configurationPath) {
+function Assembler(App, configurationPath, options) {
   if (!App)
     throw new Error("Missing app constructor");
   if (!configurationPath)
     throw new Error("Missing configuration file path");
 
+  if (!options)
+    options = {};
+
   this.App               = App;
   this.configurationPath = configurationPath;
   this.logger            = logger;
@@ -22,6 +25,7 @@ function Assembler(App, configurationPath) {
   this.reportStores      = [];
   this.eventBusReceivers = [];
   this.env               = process.env.NODE_ENV || "development";
+  this.replaying         = !!options.replaying;
 }
 
 // Assemble an app, load the reporters.
@@ -30,8 +34,9 @@ Assembler.prototype.loadCompleteApp = function (options, callback) {
 
   if (!callback) {
     callback = options;
-    options  = {}
+    options  = {};
   }
+
   if (typeof options.clearQueues == "undefined")
     options.clearQueues = false;
   if (typeof options.setupEventStore == "undefined")
@@ -40,10 +45,12 @@ Assembler.prototype.loadCompleteApp = function (options, callback) {
     options.setupReportStores = false;
   if (typeof options.startReporters == "undefined")
     options.startReporters = true;
+  if (typeof options.readOnly == "undefined")
+    options.readOnly = false;
 
   async.series([
     function (next) {
-      self.assembleApp(next);
+      self.assembleApp(options, next);
     },
     function (next) {
       if (options.clearQueues)
@@ -81,7 +88,7 @@ Assembler.prototype.loadCompleteApp = function (options, callback) {
 };
 
 // Assemble an app only, no reporters
-Assembler.prototype.assembleApp = function (callback) {
+Assembler.prototype.assembleApp = function (options, callback) {
   var self = this;
 
   if (self.app)
@@ -100,21 +107,24 @@ Assembler.prototype.assembleApp = function (callback) {
     }
   }
 
-  var createApp = function createApp() {
+  var createApp = function createApp(callback) {
     self.app = new self.App(appOptions);
     callback();
   };
 
   if (self.configuration.domain) {
-    self._assembleDomainInfrastructure(function(err){
+    self._assembleDomainInfrastructure(options, function(err){
       if (err)
         return callback(err);
       appOptions.domainRepository = self.domainRepository;
       appOptions.commandBus       = self.commandBus;
-      createApp();
+      createApp(function(){
+        DomainObject.initializeConstructors();
+        callback();
+      });
     });
   } else {
-    createApp();
+    createApp(callback);
   }
 };
 
@@ -141,7 +151,8 @@ Assembler.prototype.loadReporters = function (options, callback) {
     var eventBusReceiverConfig = {
       queueName: reporterSettings.eventBusReceiver.queue,
       logger: self.logger,
-      scope: self.configuration.scope
+      scope: self.configuration.scope,
+      replaying: self.replaying
     };
 
     if (reporterSettings.eventBusReceiver.options) {
@@ -154,7 +165,8 @@ Assembler.prototype.loadReporters = function (options, callback) {
     var reporterOptions = {
       eventBusReceiver: eventBusReceiver,
       logger: logger,
-      app: self.app
+      app: self.app,
+      replaying: self.replaying
     };
 
     var reporter = new reporterSettings.ctor(reporterOptions);
@@ -189,7 +201,8 @@ Assembler.prototype.initializeReports = function (callback) {
     var ReportStore = reportSettings.store.ctor;
     var reportStore = new ReportStore({
       uri: reportSettings.store.uri,
-      logger: self.logger
+      logger: self.logger,
+      replaying: self.replaying
     });
     Report.initialize({store: reportStore, logger: self.logger});
     self.reports.push(Report);
@@ -241,7 +254,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 +289,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 +312,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);
 };
@@ -351,7 +363,7 @@ Assembler.prototype.unloadEventBusReceivers = function (callback) {
   queue.push(self.eventBusReceivers);
 };
 
-Assembler.prototype._assembleDomainInfrastructure = function (callback) {
+Assembler.prototype._assembleDomainInfrastructure = function (options, callback) {
   var self = this;
 
   var EventStore       = self.configuration.domain.eventStore.ctor;
@@ -360,12 +372,13 @@ Assembler.prototype._assembleDomainInfrastructure = function (callback) {
     logger: self.logger
   });
 
-  function initializeDomainInfrastructure(callback) {
+  function initializeDomainInfrastructure(options, callback) {
     var EventBusEmitter  = self.configuration.domain.eventBusEmitter.ctor;
 
     var eventBusEmitterOptions = {
       logger: self.logger,
-      scope: self.configuration.scope
+      scope: self.configuration.scope,
+      replaying: self.replaying
     };
 
     if (self.configuration.domain.eventBusEmitter.options) {
@@ -379,12 +392,19 @@ Assembler.prototype._assembleDomainInfrastructure = function (callback) {
     var domainRepository = new DomainRepository({
       store: eventStore,
       emitter: eventBusEmitter,
-      logger: self.logger
+      logger: self.logger,
+      replaying: self.replaying
     });
+
+    var port;
+    if (!options.readOnly)
+      port = self.configuration.domain.commandBus && self.configuration.domain.commandBus.port;
+
     var commandBus = new CommandBus({
-      domainRepository: domainRepository,
-      port: self.configuration.domain.commandBus && self.configuration.domain.commandBus.port,
-      logger: self.logger
+      domainRepository : domainRepository,
+      port             : port,
+      logger           : self.logger,
+      replaying        : self.replaying
     });
 
     // These initializations take place on the constructor, not instances.
@@ -411,8 +431,6 @@ Assembler.prototype._assembleDomainInfrastructure = function (callback) {
       logger:           self.logger
     });
 
-    DomainObject.initializeConstructors();
-
     self.eventBusEmitter  = eventBusEmitter;
     self.domainRepository = domainRepository;
     self.commandBus       = commandBus;
@@ -425,10 +443,10 @@ Assembler.prototype._assembleDomainInfrastructure = function (callback) {
     eventStore.initialize(function(err){
       if (err)
         return callback(err);
-      initializeDomainInfrastructure(callback);
+      initializeDomainInfrastructure(options, callback);
     });
   } else {
-    initializeDomainInfrastructure(callback);
+    initializeDomainInfrastructure(options, callback);
   }
 };
 

lib/command_bus.coffee

@@ -4,19 +4,19 @@ Profiler         = require "./profiler"
 
 class CommandBus
 
-  constructor: ({@domainRepository, @logger, @port}) ->
+  constructor: ({@domainRepository, @logger, @port, @replaying}) ->
     throw new Error "Missing domain repository" unless @domainRepository?
     throw new Error "Missing logger" unless @logger?
 
-    if @port
+    if @port && !@replaying
       @server = new CommandBusServer commandBus: @, port: @port, logger: @logger
       @server.listen @port
 
     @commandHandlers = {}
 
   registerCommandHandler: (commandHandler) ->
     commandName = commandHandler.getCommandName()
-    throw new Error "A command and its handler for command named \"#{command}\" were already registered" if @commandHandlers[commandName]?
+    throw new Error "A command and its handler for command named \"#{commandName}\" were already registered" if @commandHandlers[commandName]?
     @commandHandlers[commandName] = commandHandler
 
   createNewUid: (callback) ->
@@ -48,6 +48,7 @@ class CommandBus
             commandHandler.validate (err, result) ->
               p1.end()
               if err?
+                logger.warning "executeCommand", "validation error", err
                 callback err
                 done err
                 return
@@ -57,6 +58,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,10 @@ class CommandBusServer
 
       logger.log "CommandBusServer", "deserialize command \"#{commandName}\""
       @commandBus.deserializeCommand commandName, payload, (err, command) =>
-        logger.log "CommandBusServer", "start command \"#{commandName}\""
+        if err?
+          logger.error "CommandBusServer", "error while deserializing command", err
+          return djump res, 500, error: err
+
         @commandBus.executeCommand command, (err) ->
           if err?
             logger.alert "CommandBusServer", "error while executing command (#{err})"
@@ -104,4 +107,4 @@ djump = (res, code, obj) ->
   else
     res.end()
 
-module.exports = CommandBusServer
+module.exports = CommandBusServer

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