From dd73e85b1c710b0929d2004a1d2e4cdf1500efc1 Mon Sep 17 00:00:00 2001 From: Luis Fernando Molina Date: Sat, 9 Nov 2019 19:17:54 +0100 Subject: [PATCH] Process Execution Events Implemented --- DebugLog.js | 4 + IndicatorBotProcessMainLoop.js | 233 ++++++++++++++++++++++++++++++++- ProcessExecutionEvents.js | 154 ++++++++++++++++++++++ Project.njsproj | 3 + Root.js | 7 + SensorBotProcessMainLoop.js | 231 +++++++++++++++++++++++++++++++- StatusReport.js | 38 +----- server.js | 4 +- 8 files changed, 626 insertions(+), 48 deletions(-) create mode 100644 ProcessExecutionEvents.js diff --git a/DebugLog.js b/DebugLog.js index 06b40a94e..87137a604 100644 --- a/DebugLog.js +++ b/DebugLog.js @@ -126,6 +126,10 @@ exports.newDebugLog = function newDebugLog() { if (thisObject.bot === undefined) { return; } + if (pModule === "") { // For debugging purposes + console.log(pMessage) + } + let newDate = new Date(); newDate = newDate.toISOString(); diff --git a/IndicatorBotProcessMainLoop.js b/IndicatorBotProcessMainLoop.js index bd1bd766f..ef88f2f6e 100644 --- a/IndicatorBotProcessMainLoop.js +++ b/IndicatorBotProcessMainLoop.js @@ -129,6 +129,7 @@ const STATUS_DEPENDENCIES = require(ROOT_DIR + 'StatusDependencies'); const DATA_DEPENDENCIES = require(ROOT_DIR + 'DataDependencies'); const DATA_SET = require(ROOT_DIR + 'DataSet'); + const PROCESS_EXECUTION_EVENTS = require(ROOT_DIR + 'ProcessExecutionEvents'); /* We define the datetime for the process that we are running now. This will be the official processing time for both the infraestructure and the bot. */ @@ -141,6 +142,7 @@ /* We will prepare first the infraestructure needed for the bot to run. There are 3 modules we need to sucessfullly initialize first. */ + let processExecutionEvents let userBot; let processFramework; let statusDependencies; @@ -148,7 +150,155 @@ let nextWaitTime; - initializeStatusDependencies(); + initializeProcessExecutionEvents(); + + function initializeProcessExecutionEvents() { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> initializeProcessExecutionEvents -> Entering function."); } + + processExecutionEvents = PROCESS_EXECUTION_EVENTS.newProcessExecutionEvents(bot, parentLogger) + + processExecutionEvents.initialize(pMonth, pYear, onInizialized); + + function onInizialized(err) { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Entering function."); } + + switch (err.result) { + case global.DEFAULT_OK_RESPONSE.result: { + logger.write(MODULE_NAME, "[INFO] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Execution finished well."); + startProcessExecutionEvents() + return; + } + case global.DEFAULT_RETRY_RESPONSE.result: { // Something bad happened, but if we retry in a while it might go through the next time. + logger.write(MODULE_NAME, "[WARN] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Retry Later. Requesting Execution Retry."); + nextWaitTime = 'Retry'; + loopControl(nextWaitTime); + return; + } + case global.DEFAULT_FAIL_RESPONSE.result: { // This is an unexpected exception that we do not know how to handle. + processStopped() + logger.write(MODULE_NAME, "[ERROR] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Operation Failed. Aborting the process."); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + return; + } + default: { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Unhandled err.result received. -> err.result = " + err.result); + logger.write(MODULE_NAME, "[ERROR] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Unhandled err.result received. -> err = " + err.message); + + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + return; + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> initializeProcessExecutionEvents -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } + + function startProcessExecutionEvents() { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> startProcessExecutionEvents -> Entering function."); } + + processExecutionEvents.start(onStarted); + + function onStarted(err) { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> startProcessExecutionEvents -> onStarted -> Entering function."); } + + switch (err.result) { + case global.DEFAULT_OK_RESPONSE.result: { + logger.write(MODULE_NAME, "[INFO] run -> loop -> startProcessExecutionEvents -> onStarted -> Execution finished well."); + initializeStatusDependencies(); + return; + } + case global.DEFAULT_RETRY_RESPONSE.result: { // Something bad happened, but if we retry in a while it might go through the next time. + logger.write(MODULE_NAME, "[WARN] run -> loop -> startProcessExecutionEvents -> onStarted -> Retry Later. Requesting Execution Retry."); + nextWaitTime = 'Retry'; + loopControl(nextWaitTime); + return; + } + case global.DEFAULT_FAIL_RESPONSE.result: { // This is an unexpected exception that we do not know how to handle. + processStopped() + logger.write(MODULE_NAME, "[ERROR] run -> loop -> startProcessExecutionEvents -> onStarted -> Operation Failed. Aborting the process."); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + return; + } + default: { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> startProcessExecutionEvents -> onStarted -> Unhandled err.result received. -> err.result = " + err.result); + logger.write(MODULE_NAME, "[ERROR] run -> loop -> startProcessExecutionEvents -> onStarted -> Unhandled err.result received. -> err = " + err.message); + + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + return; + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> startProcessExecutionEvents -> onStarted -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> startProcessExecutionEvents -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } function initializeStatusDependencies() { @@ -448,8 +598,7 @@ switch (err.result) { case global.DEFAULT_OK_RESPONSE.result: { logger.write(MODULE_NAME, "[INFO] run -> loop -> startUserBot -> onFinished -> Execution finished well."); - nextWaitTime = 'Normal'; - loopControl(nextWaitTime); + finishProcessExecutionEvents() return; } case global.DEFAULT_RETRY_RESPONSE.result: { // Something bad happened, but if we retry in a while it might go through the next time. @@ -681,8 +830,7 @@ switch (err.result) { case global.DEFAULT_OK_RESPONSE.result: { logger.write(MODULE_NAME, "[INFO] run -> loop -> startProcessFramework -> onFinished -> Execution finished well."); - nextWaitTime = 'Normal'; - loopControl(nextWaitTime); + finishProcessExecutionEvents() return; } case global.DEFAULT_RETRY_RESPONSE.result: { // Something bad happened, but if we retry in a while it might go through the next time. @@ -795,6 +943,81 @@ } } + function finishProcessExecutionEvents() { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> Entering function."); } + + processExecutionEvents.finish(onFinished); + + function onFinished(err) { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Entering function."); } + + switch (err.result) { + case global.DEFAULT_OK_RESPONSE.result: { + logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Execution finished well."); + nextWaitTime = 'Normal'; + loopControl(nextWaitTime); + return; + } + case global.DEFAULT_RETRY_RESPONSE.result: { // Something bad happened, but if we retry in a while it might go through the next time. + logger.write(MODULE_NAME, "[WARN] run -> loop -> finishProcessExecutionEvents -> onFinished -> Retry Later. Requesting Execution Retry."); + nextWaitTime = 'Retry'; + loopControl(nextWaitTime); + return; + } + case global.DEFAULT_FAIL_RESPONSE.result: { // This is an unexpected exception that we do not know how to handle. + processStopped() + logger.write(MODULE_NAME, "[ERROR] run -> loop -> finishProcessExecutionEvents -> onFinished -> Operation Failed. Aborting the process."); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + return; + } + default: { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> finishProcessExecutionEvents -> onFinished -> Unhandled err.result received. -> err.result = " + err.result); + logger.write(MODULE_NAME, "[ERROR] run -> loop -> finishProcessExecutionEvents -> onFinished -> Unhandled err.result received. -> err = " + err.message); + + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + return; + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> finishProcessExecutionEvents -> onFinished -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> finishProcessExecutionEvents -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } + + function loopControl(nextWaitTime) { if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> loopControl -> Entering function."); } diff --git a/ProcessExecutionEvents.js b/ProcessExecutionEvents.js new file mode 100644 index 000000000..0975a490b --- /dev/null +++ b/ProcessExecutionEvents.js @@ -0,0 +1,154 @@ +exports.newProcessExecutionEvents = function newProcessExecutionEvents(BOT, logger) { + + /* + + Here we manage the process execution events. This allow us to emit events when the process starts and finishes. Also to wait for a dependent event + of another process. + + */ + + const MODULE_NAME = "Process Execution Events"; + + let bot = BOT; + + let thisObject = { + initialize, initialize, + start: start, + finish: finish + }; + + let processThisDependsOn; // This is the process this process needs to wait for it to finishes in order to start. + + let month; + let year; + let currentProcessKey + + return thisObject; + + function initialize(pMonth, pYear, callBackFunction) { + + try { + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> Entering function."); } + + month = pMonth; + year = pYear; + + let name = 'Not Depends on any Process' + + if (bot.processNode.referenceParent !== undefined) { + + currentProcessKey = bot.processNode.referenceParent.name + "-" + bot.processNode.referenceParent.type + "-" + bot.processNode.referenceParent.id + + if (bot.processNode.referenceParent.executionStartedEvent !== undefined) { + if (bot.processNode.referenceParent.executionStartedEvent.referenceParent !== undefined) { + if (bot.processNode.referenceParent.executionStartedEvent.referenceParent.parentNode !== undefined) { + processThisDependsOn = bot.processNode.referenceParent.executionStartedEvent.referenceParent.parentNode + + name = processThisDependsOn.name + if (processThisDependsOn.code.monthlyInstances === true) { + logger.fileName = MODULE_NAME + "." + name + "." + year + "." + month; + } + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> " + currentProcessKey + "Month: " + month + " Year: " + year + " depends on " + name); } + } + } + } + } + + logger.fileName = MODULE_NAME + "." + name; + callBackFunction(global.DEFAULT_OK_RESPONSE); + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] initialize -> err = " + err.stack); + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + } + } + + function start(callBackFunction) { + + try { + + if (processThisDependsOn !== undefined) { + + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> waitForDependantProcess -> Entering Code Block."); } + + /* This forces this process to wait until the process that this one depends on, emits its event signaling that the process execution has finished. */ + + let extraCallerId = '' + if (month) { extraCallerId = extraCallerId + '-' + month } + if (year) { extraCallerId = extraCallerId + '-' + year } + + let key = processThisDependsOn.name + "-" + processThisDependsOn.type + "-" + processThisDependsOn.id + let callerId = bot.devTeam + "-" + bot.codeName + "-" + bot.process + extraCallerId + + let subscriptionId + + global.SYSTEM_EVENT_HANDLER.listenToEvent(key, 'Process Execution Finished', undefined, callerId, responseCallBack, eventsCallBack) + + function responseCallBack(message) { + if (message.result !== global.DEFAULT_OK_RESPONSE.result) { + logger.write(MODULE_NAME, "[ERROR] start -> responseCallBack -> message = " + JSON.stringify(message)) + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + } else { + subscriptionId = message.eventSubscriptionId + } + } + + function eventsCallBack(message) { + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> waitForDependantProcess -> eventsCallBack -> Entering function."); } + + /* We continue the normal flow after we learn the dependent process has finished its execution. */ + global.SYSTEM_EVENT_HANDLER.stopListening(key, 'Process Execution Finished', subscriptionId) + + if (message.event.err.result === global.DEFAULT_OK_RESPONSE.result) { + /* We emit our own event that the process started. */ + let event = { + err: global.DEFAULT_OK_RESPONSE + } + global.SYSTEM_EVENT_HANDLER.raiseEvent(currentProcessKey, 'Process Execution Started', event) + + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> eventsCallBack -> " + currentProcessKey + " Process Execution Started because " + key + " Month: " + month + " Year: " + year + " Finished."); } + } + + callBackFunction(message.event.err); + } + } else { + /* In this case, the Process does not depends on a process which needs to wait for. */ + /* We emit our own event that the process started. */ + let event = { + err: global.DEFAULT_OK_RESPONSE + } + global.SYSTEM_EVENT_HANDLER.raiseEvent(currentProcessKey, 'Process Execution Started', event) + + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> waitForDependantProcess -> " + currentProcessKey + " Process Execution Started Month: " + month + " Year: " + year); } + + callBackFunction(global.DEFAULT_OK_RESPONSE); + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] start -> err = " + err.stack); + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + } + } + + function finish(callBackFunction) { + + try { + + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] finish -> Entering function."); } + + let event = { + err: global.DEFAULT_OK_RESPONSE + } + + global.SYSTEM_EVENT_HANDLER.raiseEvent(currentProcessKey, 'Process Execution Finished', event) + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] finish -> " + currentProcessKey + " Process Execution Finished Month: " + month + " Year: " + year); } + + callBackFunction(global.DEFAULT_OK_RESPONSE); + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] finish -> err = " + err.stack); + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + } + } + +}; diff --git a/Project.njsproj b/Project.njsproj index 69959a255..d6b366b77 100644 --- a/Project.njsproj +++ b/Project.njsproj @@ -45,6 +45,9 @@ Code + + Code + Code diff --git a/Root.js b/Root.js index 59e2e34e5..707c94093 100644 --- a/Root.js +++ b/Root.js @@ -92,6 +92,13 @@ logErrors: true, logContent: false, intensiveLogging: false + }, + "Process Execution Events": { + logInfo: true, + logWarnings: false, + logErrors: true, + logContent: false, + intensiveLogging: false } }; diff --git a/SensorBotProcessMainLoop.js b/SensorBotProcessMainLoop.js index 3d20a878d..cd73954fd 100644 --- a/SensorBotProcessMainLoop.js +++ b/SensorBotProcessMainLoop.js @@ -130,6 +130,7 @@ const STATUS_REPORT = require(ROOT_DIR + 'StatusReport'); const EXCHANGE_API = require('@superalgos/exchange-gateway'); const STATUS_DEPENDENCIES = require(ROOT_DIR + 'StatusDependencies'); + const PROCESS_EXECUTION_EVENTS = require(ROOT_DIR + 'ProcessExecutionEvents'); /* We define the datetime for the process that we are running now. This will be the official processing time for both the infraestructure and the bot. */ @@ -140,15 +141,164 @@ console.log(new Date().toISOString() + " " + pad(bot.codeName, 20) + " " + pad(bot.process, 30) + " " + pad(pMonth, 2) + "/" + pad(pYear, 4) + " Entered into Main Loop # " + pad(Number(bot.loopCounter), 8) + " bot.processDatetime = " + bot.processDatetime.toISOString()); - /* We will prepare first the infraestructure needed for the bot to run. There are 3 modules we need to sucessfullly initialize first. */ + /* We will prepare first the infraestructure needed for the bot to run. There are 4 modules we need to sucessfullly initialize first. */ + let processExecutionEvents let userBot; let statusDependencies; let exchangeAPI; let nextWaitTime; - initializeStatusDependencies(); + initializeProcessExecutionEvents(); + + function initializeProcessExecutionEvents() { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> initializeProcessExecutionEvents -> Entering function."); } + + processExecutionEvents = PROCESS_EXECUTION_EVENTS.newProcessExecutionEvents(bot, parentLogger) + + processExecutionEvents.initialize(pMonth, pYear, onInizialized); + + function onInizialized(err) { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Entering function."); } + + switch (err.result) { + case global.DEFAULT_OK_RESPONSE.result: { + logger.write(MODULE_NAME, "[INFO] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Execution finished well."); + startProcessExecutionEvents() + return; + } + case global.DEFAULT_RETRY_RESPONSE.result: { // Something bad happened, but if we retry in a while it might go through the next time. + logger.write(MODULE_NAME, "[WARN] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Retry Later. Requesting Execution Retry."); + nextWaitTime = 'Retry'; + loopControl(nextWaitTime); + return; + } + case global.DEFAULT_FAIL_RESPONSE.result: { // This is an unexpected exception that we do not know how to handle. + processStopped() + logger.write(MODULE_NAME, "[ERROR] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Operation Failed. Aborting the process."); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + return; + } + default: { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Unhandled err.result received. -> err.result = " + err.result); + logger.write(MODULE_NAME, "[ERROR] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> Unhandled err.result received. -> err = " + err.message); + + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + return; + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> initializeProcessExecutionEvents -> onInizialized -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> initializeProcessExecutionEvents -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } + + function startProcessExecutionEvents() { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> startProcessExecutionEvents -> Entering function."); } + + processExecutionEvents.start(onStarted); + + function onStarted(err) { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> startProcessExecutionEvents -> onStarted -> Entering function."); } + + switch (err.result) { + case global.DEFAULT_OK_RESPONSE.result: { + logger.write(MODULE_NAME, "[INFO] run -> loop -> startProcessExecutionEvents -> onStarted -> Execution finished well."); + initializeStatusDependencies(); + return; + } + case global.DEFAULT_RETRY_RESPONSE.result: { // Something bad happened, but if we retry in a while it might go through the next time. + logger.write(MODULE_NAME, "[WARN] run -> loop -> startProcessExecutionEvents -> onStarted -> Retry Later. Requesting Execution Retry."); + nextWaitTime = 'Retry'; + loopControl(nextWaitTime); + return; + } + case global.DEFAULT_FAIL_RESPONSE.result: { // This is an unexpected exception that we do not know how to handle. + processStopped() + logger.write(MODULE_NAME, "[ERROR] run -> loop -> startProcessExecutionEvents -> onStarted -> Operation Failed. Aborting the process."); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + return; + } + default: { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> startProcessExecutionEvents -> onStarted -> Unhandled err.result received. -> err.result = " + err.result); + logger.write(MODULE_NAME, "[ERROR] run -> loop -> startProcessExecutionEvents -> onStarted -> Unhandled err.result received. -> err = " + err.message); + + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + return; + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> startProcessExecutionEvents -> onStarted -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> startProcessExecutionEvents -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } function initializeStatusDependencies() { @@ -427,8 +577,7 @@ switch (err.result) { case global.DEFAULT_OK_RESPONSE.result: { logger.write(MODULE_NAME, "[INFO] run -> loop -> startUserBot -> onFinished > Execution finished well."); - nextWaitTime = 'Normal'; - loopControl(nextWaitTime); + finishProcessExecutionEvents() return; } case global.DEFAULT_RETRY_RESPONSE.result: { // Something bad happened, but if we retry in a while it might go through the next time. @@ -541,6 +690,80 @@ } } + function finishProcessExecutionEvents() { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> Entering function."); } + + processExecutionEvents.finish(onFinished); + + function onFinished(err) { + + try { + + if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Entering function."); } + + switch (err.result) { + case global.DEFAULT_OK_RESPONSE.result: { + logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Execution finished well."); + nextWaitTime = 'Normal'; + loopControl(nextWaitTime); + return; + } + case global.DEFAULT_RETRY_RESPONSE.result: { // Something bad happened, but if we retry in a while it might go through the next time. + logger.write(MODULE_NAME, "[WARN] run -> loop -> finishProcessExecutionEvents -> onFinished -> Retry Later. Requesting Execution Retry."); + nextWaitTime = 'Retry'; + loopControl(nextWaitTime); + return; + } + case global.DEFAULT_FAIL_RESPONSE.result: { // This is an unexpected exception that we do not know how to handle. + processStopped() + logger.write(MODULE_NAME, "[ERROR] run -> loop -> finishProcessExecutionEvents -> onFinished -> Operation Failed. Aborting the process."); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + return; + } + default: { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> finishProcessExecutionEvents -> onFinished -> Unhandled err.result received. -> err.result = " + err.result); + logger.write(MODULE_NAME, "[ERROR] run -> loop -> finishProcessExecutionEvents -> onFinished -> Unhandled err.result received. -> err = " + err.message); + + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + return; + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> finishProcessExecutionEvents -> onFinished -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] run -> loop -> finishProcessExecutionEvents -> err = " + err.stack); + logger.persist(); + clearInterval(fixedTimeLoopIntervalHandle); + clearTimeout(nextLoopTimeoutHandle); + clearTimeout(checkLoopHealthHandle); + bot.enableCheckLoopHealth = false; + callBackFunction(err); + } + } + function loopControl(nextWaitTime) { if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> loopControl -> Entering function."); } diff --git a/StatusReport.js b/StatusReport.js index f239e826a..7289fb35a 100644 --- a/StatusReport.js +++ b/StatusReport.js @@ -59,43 +59,7 @@ timePath = "/" + year + "/" + month; } - if (owner.waitUntilNextUpdate === true) { - - if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> waitUntilNextUpdate -> Entering Code Block."); } - - /* This forces this process to wait until the process that this one depends on, updates its status report. */ - - let extraCallerId = '' - if (pMonth) { extraCallerId = extraCallerId + '-' + pMonth } - if (pYear) { extraCallerId = extraCallerId + '-' + pYear } - - let key = owner.devTeam + "-" + owner.bot + "-" + owner.process - let callerId = bot.devTeam + "-" + bot.codeName + "-" + bot.process + extraCallerId - - let subscriptionIdStatusReport - - global.SYSTEM_EVENT_HANDLER.listenToEvent(key, 'Status Report Updated', undefined, callerId, responseCallBack, eventsCallBack) - - function responseCallBack(message) { - if (message.result !== global.DEFAULT_OK_RESPONSE.result) { - logger.write(MODULE_NAME, "[ERROR] initialize -> responseCallBack -> message = " + JSON.stringify(message)) - callBackFunction(global.DEFAULT_FAIL_RESPONSE); - } else { - subscriptionIdStatusReport = message.eventSubscriptionId - } - } - - function eventsCallBack() { - if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> eventsCallBack -> Entering function."); } - - /* We continue the normal flow after we learn the dependent process has updated its status report. */ - global.SYSTEM_EVENT_HANDLER.stopListening(key, 'Status Report Updated', subscriptionIdStatusReport) - callBackFunction(global.DEFAULT_OK_RESPONSE); - } - } else { - /* In this case, the Status Report does not depends on a process which needs to wait for. */ - callBackFunction(global.DEFAULT_OK_RESPONSE); - } + callBackFunction(global.DEFAULT_OK_RESPONSE); } catch (err) { logger.write(MODULE_NAME, "[ERROR] initialize -> err = "+ err.stack); diff --git a/server.js b/server.js index 5a6d11e39..4c0a1ee49 100644 --- a/server.js +++ b/server.js @@ -73,8 +73,8 @@ if (global.TASK_NODE !== undefined) { } else { // I use this section to debug in standalone mode. let argument - argument = ' {"type":"Task","name":"Runs Backtests","bot":{"type":"Trading Bot Instance","code":{"team":"AAMasters","bot":"AAJason","repo":"AAJason-Trading-Bot"},"processes":[{"type":"Process Instance","subType":"Trading Process Instance","name":"Multi Period","code":{"process":"Multi-Period"},"session":{"type":"Backtesting Session","name":"Backtest 01","code":{"folderName":"Backtest-01"},"parameters":{"type":"Parameters","name":"Parameters","baseAsset":{"type":"Base Asset","name":"Base Asset","code":{"name":"BTC","initialBalance":0.001,"minimumBalance":0.0001,"maximumBalance":1},"id":"666f6f10-2f17-4f5e-951c-b532bdb574d2"},"timeRange":{"type":"Time Range","name":"Time Range","code":{"initialDatetime":"2019-09-24T00:00:00.000Z","finalDatetime":"2019-09-26T00:00:00.000Z"},"id":"172958dd-b268-4444-8163-70979538ea6e"},"timePeriod":{"type":"Time Period","name":"Time Period","code":"10-min","id":"37ee839d-9a7c-410c-9251-c33b8f1b99f7"},"slippage":{"type":"Slippage","name":"Slippage","code":{"positionRate":0.1,"stopLoss":0.1,"takeProfit":0.1},"id":"86ebd973-6ad9-4070-aece-a87623f24c2e"},"feeStructure":{"type":"Fee Structure","name":"Fee Structure","code":{"maker":0.15,"taker":0.25},"id":"a0ec3f6f-6047-4740-87fe-393b1c714adb"},"key":{"type":"Exchange Account Key","name":"New Key","code":"Paste your exchange API secret key here and the put the key name as this key object title. Secret keys are filtered out and NOT exported when using the SHARE menu option on any object at your workspace. Secret keys ARE downloaded when using the download button.","id":"368f879d-e363-47c7-b55e-7b927d896b20"},"id":"09db5e1e-a7e8-468c-ad70-0d2ab1c8b8b8"},"layerManager":{"type":"Layer Manager","name":"New Layer Manager","layers":[{"type":"Layer","name":"Live","code":{"product":"Live-Trading-History"},"id":"0ec05e1c-230b-44a1-9f2c-fba0ea2ff9e2"},{"type":"Layer","name":"Trades","code":{"product":"Simulation-Trades"},"id":"6326e03a-96c0-4d21-b877-f2c6a0419fc5"},{"type":"Layer","name":"Strategies","code":{"product":"Simulation-Strategies"},"id":"bda1e06c-e569-4bd5-95fb-90154aafb003"},{"type":"Layer","name":"Simulation","code":{"product":"Trading-Simulation"},"id":"d58d8cd2-f125-4f0c-b241-08e2af8ac4be"},{"type":"Layer","name":"Formulas & Conditions","code":{"product":"Simulation-Conditions"},"id":"15dfeca0-acd9-40d3-afe0-4805654bf0d0"}],"id":"884c4dd7-4beb-43fb-bda5-d6097d58ae09"},"socialBots":{"type":"Social Bots","name":"New Social Bots","bots":[],"id":"281eb022-cf9f-4ac2-978b-c8358bee8fec"},"id":"121202ea-6735-4bf3-8a95-aaa55f00f6e7"},"id":"3f483482-d8a1-4755-ba50-7cac6296fa5f"}],"id":"25fe9099-b53d-4757-8a63-3f8a15875f33"},"id":"5b21d2d2-0f52-4408-83eb-ddaa844ec58c"}' - + argument = ' {"type":"Task","name":"Brings Trades Records from the Exchange","bot":{"type":"Sensor Bot Instance","name":"Charly","code":{"team":"AAMasters","bot":"AACharly","repo":"AACharly-Sensor-Bot"},"processes":[{"type":"Process Instance","subType":"Sensor Process Instance","name":"Live Trades","code":{"process":"Live-Trades"},"id":"338266fa-14c2-42eb-b0e5-d0a8ed9309fc","referenceParent":{"type":"Process Definition","name":"Live-Trades","code":{},"processOutput":{"type":"Process Output","name":"New Process Output","outputDatasets":[],"id":"9ffdaadd-5204-4436-b89e-cabd8f0e5829"},"processDependencies":{"type":"Process Dependencies","name":"New Process Dependencies","statusDependencies":[],"dataDependencies":[],"id":"a98b94ad-0cd4-4e54-ade4-68b0ab577414"},"statusReport":{"type":"Status Report","name":"New Status Report","id":"9f3ea81f-2aa1-41ec-907f-c003dec0c827"},"executionStartedEvent":{"type":"Execution Started Event","name":"New Execution Started Event","id":"942c8337-5d22-413d-9e44-e2da9a5479f1"},"executionFinishedEvent":{"type":"Execution Finished Event","name":"New Execution Finished Event","id":"d5f1f2b3-ffa5-442b-b875-e91a786337ae","parentNode":{"type":"Process Definition","name":"Live-Trades","id":"371a2507-655b-46c5-b959-52f527cc4d7e","code":{}}},"calculations":{"type":"Calculations Procedure","name":"New Calculations Procedure","id":"e3774584-43e5-4dc5-b097-7b16476d586a"},"dataBuilding":{"type":"Data Building Procedure","name":"New Data Building Procedure","id":"0f8cc852-ad72-40fe-bf4a-866e68439492"},"id":"371a2507-655b-46c5-b959-52f527cc4d7e"}},{"type":"Process Instance","subType":"Sensor Process Instance","name":"Hole Fixing","code":{"process":"Hole-Fixing"},"id":"3d7e0bdd-64e3-4442-8742-cc9282a82631","referenceParent":{"type":"Process Definition","name":"Hole-Fixing","code":{},"processOutput":{"type":"Process Output","name":"New Process Output","outputDatasets":[{"type":"Output Dataset","name":"New Output Dataset","id":"646ec2e6-98cb-489f-abb6-98b9bbdfa5f0","referenceParent":{"type":"Dataset Definition","name":"Trades","code":{},"id":"80c4ee03-aed4-4d8e-8145-a49f8b85df04"}}],"id":"4c577394-cc99-4d9d-ae73-38f28e08405d"},"processDependencies":{"type":"Process Dependencies","name":"New Process Dependencies","statusDependencies":[{"type":"Status Dependency","name":"New Status Dependency","id":"b707a221-e8f4-4d6f-824c-8e26e92b8093","referenceParent":{"type":"Status Report","name":"New Status Report","id":"03aa3c2e-d33c-4054-9150-1b641bc02e96"}},{"type":"Status Dependency","name":"New Status Dependency","id":"e9aedbfd-8b76-44c4-bfda-733da4f59023"},{"type":"Status Dependency","name":"New Status Dependency","id":"7a3ebb1b-1d28-4abd-9875-6b0d9c2ecea8","referenceParent":{"type":"Status Report","name":"New Status Report","id":"c542a73e-a350-431f-ad3e-10f3ebef8ec7"}}],"dataDependencies":[{"type":"Data Dependency","name":"New Data Dependency","id":"cab5ab89-a3e2-4833-a532-3884f0ff6fa9","referenceParent":{"type":"Dataset Definition","name":"Trades","code":{},"id":"80c4ee03-aed4-4d8e-8145-a49f8b85df04"}}],"id":"55891378-cdfa-4b6a-895e-235987c7b568"},"statusReport":{"type":"Status Report","name":"New Status Report","id":"03aa3c2e-d33c-4054-9150-1b641bc02e96"},"executionStartedEvent":{"type":"Execution Started Event","name":"New Execution Started Event","id":"39a7e02f-1b3b-49ea-9714-49f8fccc768d"},"executionFinishedEvent":{"type":"Execution Started Event","name":"New Execution Started Event","id":"39a7e02f-1b3b-49ea-9714-49f8fccc768d","referenceParent":{"type":"Execution Finished Event","name":"New Execution Finished Event","id":"d5f1f2b3-ffa5-442b-b875-e91a786337ae","parentNode":{"type":"Process Definition","name":"Live-Trades","id":"371a2507-655b-46c5-b959-52f527cc4d7e","code":{}}}},"calculations":{"type":"Calculations Procedure","name":"New Calculations Procedure","id":"8777f6f4-8e49-487c-8a77-2f1e4be738b1"},"dataBuilding":{"type":"Data Building Procedure","name":"New Data Building Procedure","id":"0d2e6c5a-cb53-4b72-b688-db9478024182"},"id":"b9d87bfc-1df9-4be3-b8c8-41131f5a9dc4"}}],"id":"85c4f582-56ee-4b8e-95b2-aa61baedfa39"},"id":"2fcbea08-a941-4891-80fd-236eacd83c11"}' + try { global.TASK_NODE = JSON.parse(argument) } catch (err) {