Adding workflow durability by allowing persistence between each stage

lib/client.js

@@ -232,4 +232,11 @@ BPMNProcessClient.prototype.traceDatetime = function(key, value) {
  key,
  value);
  }
-};
+};
+
+/**
+ * @param {Boolean=} closeConnection
+ */
+BPMNProcessClient.prototype.persist = function(callback, closeConnection) {
+ return this._getImplementation().persist(closeConnection, callback);
+}

lib/manager.js

@@ -35,8 +35,12 @@ var ProcessManager = exports.ProcessManager = function(options){
  self._initializationError = null;
  self._definitionsToInitialize = [];
  self._initialiseCallbacks = [];
+ self._autoCleanFinishedProcess = false;
 
  self._persistency = null;
+ if (options.autoCleanFinishedProcess) {
+ self._autoCleanFinishedProcess = true;
+ }
  if (options.persistencyOptions) {
  self._persistency = new Persistency(options.persistencyOptions);
  self._doneLoadingHandler = options.persistencyOptions.doneLoading;
@@ -172,7 +176,7 @@ ProcessManager.prototype._initialiseDefinition = function(processDefinition){
  var currentDefinition = self._definitionsToInitialize.pop(); // get the next definition
 
  if(self._processDefinitions[currentDefinition.name] || // if the definition already exist it means the processes were already loaded
- !self._persistency){ // if there is no persistency nothing needs to be loaded
+ !self._persistency || self._autoCleanFinishedProcess){ // if there is no persistency nothing needs to be loaded
 
  self._processDefinitions[currentDefinition.name] = currentDefinition; // we simply add or replace the definition
  return next();
@@ -359,8 +363,21 @@ ProcessManager.prototype.get = function get(processId, callback) {
  * @param {Function} callback
  */
 ProcessManager.prototype._createSingleProcess = function(processId, processName, callback){
+ var self = this;
  createBPMNProcess(processId, this._processDefinitions[processName], this._processHandlers[processName], this._persistency, function(err, bpmnProcess){
  callback(err, bpmnProcess);
+ if (self._autoCleanFinishedProcess) {
+ bpmnProcess.on('ProcessEnded', function () {
+ var endedProcessId = this.getProcessId();
+ if (self._processCache[endedProcessId]) {
+ delete self._processCache[endedProcessId];
+ }
+ if (self._persistency != null) {
+ self._persistency.delete(endedProcessId, this.getProcessDefinition().name, function() {
+ });
+ }
+ });
+ }
  });
 };
 
@@ -483,6 +500,35 @@ ProcessManager.prototype.createProcess = function(descriptors, callback) {
  });
 };
 
+ProcessManager.prototype.resumeProcess = function(processId, callback) {
+ var self = this;
+ if (!self._processCache[processId]){
+ //Load process from db
+ self._persistency.load(processId, Object.keys(self._processDefinitions)[0], function(err, document) {
+ if (err) {
+ return callback(err);
+ } else {
+ self._createSingleProcess(document.processId, Object.keys(self._processDefinitions)[0], function(err, bpmnProcess){ // create the process
+ if(err){
+ return callback(err);
+ }
+
+ self._processCache[processId] = bpmnProcess;
+ var flowObject = bpmnProcess.processDefinition.getFlowObjectByName(bpmnProcess.state.tokens[0].position);
+ bpmnProcess._putTokenAt(flowObject, {});
+ callback(null);
+ });
+ }
+
+ });
+ } else {
+ var bpmnProcess = self._processCache[processId];
+ var flowObject = bpmnProcess.processDefinition.getFlowObjectByName(bpmnProcess.state.tokens[0].position);
+ bpmnProcess._putTokenAt(flowObject, {});
+ callback(null);
+ }
+};
+
 /**
  * @param {Function} callback
  */
@@ -640,4 +686,8 @@ ProcessManager.prototype.getDefinitionNames = function(callback){
  */
 ProcessManager.prototype.createServer = function(options, restifyOptions){
  return rest.createServer(this, options, restifyOptions);
+};
+
+ProcessManager.prototype.getNumberOfProcesses = function(){
+ return Object.keys(this._processCache).length;
 };

lib/persistency/mongodb.js

@@ -69,6 +69,15 @@ Persistency.prototype.load = function(processId, processName, done) {
  this._execute(this._find, {processId: processId, processName: processName}, done);
 };
 
+/**
+ * @param {String} processId
+ * @param {String} processName
+ * @param done
+ */
+Persistency.prototype.delete = function(processId, processName, done) {
+ this._execute(this._delete, {processId: processId, processName: processName}, done);
+};
+
 /**
  * @param {String} processName
  * @param done
@@ -232,6 +241,21 @@ Persistency.prototype._find = function(db, query, done) {
  );
 };
 
+/**
+ * @param db
+ * @param query
+ * @param done
+ * @private
+ */
+Persistency.prototype._delete = function(db, query, done) {
+ var processId = query.processId;
+ var processName = query.processName;
+ var collection = db.collection(processName);
+ collection.deleteOne({processId: processId}, function(err) {
+ done(err);
+ });
+}
+
 /**
  * @param db
  * @param processName

lib/persistency/persistency.js

@@ -51,6 +51,15 @@ Persistency.prototype.loadAll = function(processName, done) {
  this.implementation.loadAll(processName, done);
 };
 
+/**
+ * @param {String} processId
+ * @param {String} processName
+ * @param done
+ */
+Persistency.prototype.delete = function(processId, processName, done) {
+ this.implementation.delete(processId, processName, done);
+};
+
 /**
  * @param done
  */

lib/process.js

@@ -422,7 +422,7 @@ BPMNProcess.prototype.getProperties = function() {
 /**
  * @param {Boolean=} closeConnection
  */
-BPMNProcess.prototype.persist = function(closeConnection) {
+BPMNProcess.prototype.persist = function(closeConnection, persistCallback) {
  var mainProcess, doneSaving, persistentData;
  var persistency = this.persistency;
 
@@ -437,6 +437,9 @@ BPMNProcess.prototype.persist = function(closeConnection) {
  if (mainProcess.doneSavingHandler) {
  mainProcess.doneSavingHandler.call(mainProcess.processClient, error);
  }
+ if (persistCallback) {
+ persistCallback(error);
+ }
  } else {
  if (mainProcess.doneSavingHandler) {
  mainProcess.doneSavingHandler.call(mainProcess.processClient, null, savedData);
@@ -451,6 +454,10 @@ BPMNProcess.prototype.persist = function(closeConnection) {
  } else {
  mainProcess._emitDeferredEvents();
  }
+ if (persistCallback) {
+ persistCallback(null);
+ }
+
  }
  };
 
@@ -766,7 +773,9 @@ BPMNProcess.prototype.onProcessEnd = function(endEventName, isMainProcess, done)
  // emit event ?
  // pass the manager to the process ?
  // pass a function that will be called here ?
- self.persist(true);
+ self.persist(true, function() {
+ self._emitEvent("ProcessEnded", "ProcessEnded", {});
+ });
  }
  };
 

package.json

@@ -23,14 +23,17 @@
  "async": "0.9.0"
  },
  "optionalDependencies": {
- "mongodb": ">= 1.3.10",
+ "mongodb": "^2.2.31",
  "e2e-transaction-logger": "~0.1.0"
  },
  "devDependencies": {
  "nodeunit": ">= 0.7.4",
  "commander": ">= 2.0.0",
  "istanbul": ">= 0.1.44"
  },
+ "scripts": {
+ "coverage": "istanbul cover test/nodeunittraverse.js -- test/core test/mongodb test/public"
+ },
  "repository": "https://github.com/e2ebridge/bpmn.git",
  "author": "support@e2ebridge.com",
  "license": "MIT"