diff --git a/IndicatorBotProcessMainLoop.js b/IndicatorBotProcessMainLoop.js index 8a57bbf59..b1bdfc5d1 100644 --- a/IndicatorBotProcessMainLoop.js +++ b/IndicatorBotProcessMainLoop.js @@ -120,6 +120,9 @@ if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> Entering function."); } + /* Loop Heartbeat sent to the UI */ + hearBeat() + /* We define here all the modules that the rest of the infraestructure, including the bots themselves can consume. */ const UTILITIES = require(ROOT_DIR + 'CloudUtilities'); @@ -791,6 +794,10 @@ if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> loopControl -> Entering function."); } if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> loopControl -> nextWaitTime = " + nextWaitTime); } + /* We show we reached the end of the loop. */ + + hearBeat() + /* Here we check if we must stop the loop gracefully. */ shallWeStop(onStop, onContinue); @@ -916,6 +923,15 @@ callBackFunction(err); } } + + function hearBeat() { + let key = global.USER_DEFINITION.bot.processes[global.sequenceStep - 1].name + '-' + global.USER_DEFINITION.bot.processes[global.sequenceStep - 1].type + '-' + global.USER_DEFINITION.bot.processes[global.sequenceStep - 1].id + + let event = { + seconds: (new Date()).getSeconds() + } + global.SYSTEM_EVENT_HANDLER.raiseEvent(key, 'Heartbeat', event) + } } catch (err) { parentLogger.write(MODULE_NAME, "[ERROR] run -> err = "+ err.stack); clearInterval(fixedTimeLoopIntervalHandle); diff --git a/StatusReport.js b/StatusReport.js index 106dfa3a4..1c6d4e497 100644 --- a/StatusReport.js +++ b/StatusReport.js @@ -54,7 +54,76 @@ timePath = "/" + year + "/" + month; } - callBackFunction(global.DEFAULT_OK_RESPONSE); + + if (owner.waitUntilNextUpdate === true) { + /* This forces this process to wait until the process that this one depends on, updates its status report. */ + + let key = owner.devTeam + "-" + owner.bot + "-" + owner.process + let callerId = bot.devTeam + "-" + bot.codeName + "-" + bot.process + + let eventSubscriptionIdStatusReport + let eventSubscriptionIdProcessTerminated + + eventSubscriptionIdStatusReport = global.SYSTEM_EVENT_HANDLER.listenToEvent(key, 'Status Report Updated', undefined, callerId, responseCallBackStatusReport, eventsCallBackStatusReport) + + function responseCallBackStatusReport(message) { + if (message.result !== global.DEFAULT_OK_RESPONSE.result) { + logger.write(MODULE_NAME, "[ERROR] initialize -> Could not register event listener for event 'Status Report Updated' -> Message = " + message); + stopListeningAll() + let event = { + reason: 'This process depends on ' + key + ' which is currently not running.' + } + global.SYSTEM_EVENT_HANDLER.raiseEvent(key, 'Process Terminated', event) + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + } + } + + function eventsCallBackStatusReport() { + /* We continue the normal flow after we learn the dependent process has updated its status report. */ + stopListeningAll() + callBackFunction(global.DEFAULT_OK_RESPONSE); + } + + eventSubscriptionIdProcessTerminated = global.SYSTEM_EVENT_HANDLER.listenToEvent(key, 'Process Terminated', undefined, callerId, responseCallBackProcessTerminated, eventsCallBackProcessTerminated) + + function responseCallBackProcessTerminated(message) { + if (message.result !== global.DEFAULT_OK_RESPONSE.result) { + logger.write(MODULE_NAME, "[ERROR] initialize -> Could not register event listener for event 'Process Terminated' -> Message = " + message); + stopListeningAll() + let event = { + reason: 'This process depends on ' + key + ' which is currently not running.' + } + global.SYSTEM_EVENT_HANDLER.raiseEvent(key, 'Process Terminated', event) + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + } + } + + function eventsCallBackProcessTerminated() { + /* If the process we depend on to start was terminated, there is no point to continue waiting. We need to cloese everything al terminate this process too. */ + logger.write(MODULE_NAME, "[ERROR] initialize -> Received event 'Process Terminated' from a dependency we are waiting for. -> Message = " + message); + + stopListeningAll() + let event = { + reason: 'This process depends on ' + key + ' which has been terminated.' + } + global.SYSTEM_EVENT_HANDLER.raiseEvent(key, 'Process Terminated', event) + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + } + + function stopListeningAll() { + if (eventSubscriptionIdStatusReport) { + stopListening(key, 'Status Report Updated', eventSubscriptionIdStatusReport) + } + if (eventSubscriptionIdProcessTerminated) { + stopListening(key, 'Process Terminated', eventSubscriptionIdStatusReport) + } + } + + } else { + /* In this case, the Status Report does not depends on a process which needs to wait for. */ + callBackFunction(global.DEFAULT_OK_RESPONSE); + } + } catch (err) { logger.write(MODULE_NAME, "[ERROR] initialize -> err = "+ err.stack); callBackFunction(global.DEFAULT_FAIL_RESPONSE); @@ -199,6 +268,10 @@ logger.write(MODULE_NAME, "[INFO] save -> onFileCreated -> Content written = " + fileContent); } + /* Here we raise the event stating that this status report was updated. */ + let key = bot.devTeam + "-" + bot.codeName + "-" + bot.process + global.SYSTEM_EVENT_HANDLER.raiseEvent(key, 'Status Report Updated') + callBackFunction(global.DEFAULT_OK_RESPONSE); return; } diff --git a/TradingEngineProcessMainLoop.js b/TradingEngineProcessMainLoop.js index c1a6a7ecc..a8df4efef 100644 --- a/TradingEngineProcessMainLoop.js +++ b/TradingEngineProcessMainLoop.js @@ -125,6 +125,9 @@ if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> Entering function."); } + /* Loop Heartbeat sent to the UI */ + hearBeat() + /* We define here all the modules that the rest of the infraestructure, including the bots themselves can consume. */ const UTILITIES = require(ROOT_DIR + 'CloudUtilities'); @@ -870,6 +873,10 @@ if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> loopControl -> Entering function."); } if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> loopControl -> nextWaitTime = " + nextWaitTime); } + /* We show we reached the end of the loop. */ + + hearBeat() + /* Here we check if we must stop the loop gracefully. */ shallWeStop(onStop, onContinue); @@ -996,6 +1003,15 @@ callBackFunction(err); } } + + function hearBeat() { + let key = global.USER_DEFINITION.bot.processes[global.sequenceStep - 1].name + '-' + global.USER_DEFINITION.bot.processes[global.sequenceStep - 1].type + '-' + global.USER_DEFINITION.bot.processes[global.sequenceStep - 1].id + + let event = { + seconds: (new Date()).getSeconds() + } + global.SYSTEM_EVENT_HANDLER.raiseEvent(key, 'Heartbeat', event) + } } catch (err) { parentLogger.write(MODULE_NAME, "[ERROR] run -> err = "+ err.stack); clearInterval(fixedTimeLoopIntervalHandle); diff --git a/server.js b/server.js index 424e895cf..95a2b25f8 100644 --- a/server.js +++ b/server.js @@ -78,11 +78,11 @@ else { // I use this section to debug in standalone mode. const EVENT_HANDLER_MODULE = require('../Libraries/SystemEventsClient/SystemEventHandler.js'); global.SYSTEM_EVENT_HANDLER = EVENT_HANDLER_MODULE.newSystemEventHandler() -global.SYSTEM_EVENT_HANDLER.initialize('Clone Executor', bootLoader) +global.SYSTEM_EVENT_HANDLER.initialize('Task Server', bootLoader) function bootLoader() { - global.SYSTEM_EVENT_HANDLER.createEventHandler(global.USER_DEFINITION.name) + //global.SYSTEM_EVENT_HANDLER.createEventHandler(global.USER_DEFINITION.name) /* tryToListenToCockpit() @@ -101,7 +101,14 @@ function bootLoader() { for (let i = 0; i < global.USER_DEFINITION.bot.processes.length; i++) { let code = global.USER_DEFINITION.bot.processes[i].code + + /* Add to the execution sequence list. */ sequenceList.push(code) + + /* Create the event handler for each process. This event handlers are where the status reports updated events are raised. */ + + let key = code.devTeam + "-" + code.bot + "-" + code.process + global.SYSTEM_EVENT_HANDLER.createEventHandler(key) } /* Heartbeat sent to the UI */ @@ -153,12 +160,22 @@ if (process.env.RUN_SEQUENCE !== undefined) { function startSequence () { if (isRunSequence) { process.on('message', message => { - if (message === 'STOP') { + if (message === 'Stop this Task') { runClonExecutor = false; /* Cleaning Before Exiting. */ clearInterval(heartBeatInterval) + for (let i = 0; i < global.USER_DEFINITION.bot.processes.length; i++) { + let code = global.USER_DEFINITION.bot.processes[i].code + + /* Delete the event handler for each process. */ + + let key = code.devTeam + "-" + code.codeName + "-" + code.process + global.SYSTEM_EVENT_HANDLER.raiseEvent(key, 'Process Terminated') + global.SYSTEM_EVENT_HANDLER.deleteEventHandler(key) + } + global.SYSTEM_EVENT_HANDLER.finalize() global.SYSTEM_EVENT_HANDLER = undefined console.log("[INFO] CloneExecutor -> Clone Executor Stopped.");