Updates required for persisting function inputs in the persistence log

Author: balis

engine2/index.js

@@ -216,7 +216,7 @@ Engine.prototype.emitSignals = function(sigs, cb) {
                 } 
 
                 // signals which don't have "source" are workflow inputs ==> need to be persisted
-                // FIXME: add a flag to check if persistence is enabled
+                // FIXME: add a flag to check if persistence is enabled. EDIT: no need for flag, this is checked in ONE PLACE in 'hflow'
                 if (!s.source) {
                     engine.eventServer.emit("persist", ["input", +engine.wfId, s]);
                 }

engine2/process.js

@@ -316,25 +316,41 @@ var ProcLogic = function() {
                 funcOuts, emul,
                 proc.engine.eventServer,
                 proc.engine.config, 
-                function(err, outs, options) {
+                function(err, ins, outs, options) { // 'ins' is returned only for the purpose of persistence
 		    //onsole.log("FUNC INVOKED");
 		    //onsole.log("INS: ", JSON.stringify(proc.sigValues, null, 2));
 		    //onsole.log("OUTS: ", outs);
-                    var outsArray = [];
+
+                    // convert ins and outs object-arrays to arrays for the purpose of persistence
+                    var outsArray = [], insArray = [];
                     if (outs == null) {
                        outsArray = null; 
                     } else {
-                        outs.forEach(function(out) {
+                        outs.forEach(function(out, i) {
                             outsArray.push(out);
+                            // FIXME: the code below is repeated below in "postInvoke"
+                            // --> TODO: make a function for adding metadata to a signal 
+                            outsArray[i]["_id"] = +funcOuts[i];
+                            outsArray[i]["source"] = +proc.procId;
+                            outsArray[i]["firingId"] = +proc.firingId;
+                        });        
+                    }                    
+                    if (ins == null) {
+                       insArray = null; 
+                    } else {
+                        ins.forEach(function(input) {
+                            insArray.push(input);
                         });
                     }
 
+
                     // persist outputs (originally persistence was disabled DURING recovery, 
                     // but it's probably wrong: currently even when recovering from a previous log, 
-                    // workflow execution is persisted normally to a new log
+                    // workflow execution is persisted normally to a new log)
                     //if (!options || !options.recovered) {
                         proc.engine.eventServer.emit("persist", 
-                            ["fired", proc.appId, proc.procId, proc.firingId, outsArray]);
+                            // FIXME: insArray only added for testing "pregel" algorithm
+                            ["fired", proc.appId, proc.procId, proc.firingId, outsArray, insArray]);
                     //}
                     err ? cb(err): cb(null, outs, asyncInvocation, funcIns, funcOuts);
                 }

wflib/index.js

@@ -1479,14 +1479,14 @@ function public_invokeProcFunction(wfId, procId, firingId, insIds_, insValues, o
         prepareFuncOutputs(function(outs, recovered) {
             if (emulate) {
                 return setTimeout(function() {
-                    cb(null, outs);
+                    cb(null, ins, outs);
                 }, 100);
             }
 
             // when this is a recovered firing, unless the process has set flag "executeWhenRecovering",
             // we don't invoke the function, just immediately return the recovered outputs!
             if (recovered && !procInfo.executeWhenRecovering) {
-                return cb(null, outs, {"recovered": "true"});
+                return cb(null, ins, outs, {"recovered": "true"});
             }
 
             if ((procInfo.fun == "null") || (!procInfo.fun)) {
@@ -1566,7 +1566,7 @@ function public_invokeProcFunction(wfId, procId, firingId, insIds_, insValues, o
                             options.recovered = true;
                         }
                     }
-                    cb(null, outs, options);
+                    cb(null, ins, outs, options);
                 });
             });
         });