diff --git a/DataDependencies.js b/DataDependencies.js index a833ad6a3..4f646a80c 100644 --- a/DataDependencies.js +++ b/DataDependencies.js @@ -10,7 +10,8 @@ nodeArray: undefined, dataSetsModulesArray: [], isItADepenency: isItADepenency, - initialize: initialize + initialize: initialize, + finalize: finalize }; let filter = new Map() @@ -103,6 +104,16 @@ } } + function finalize() { + for (let i = 0; i < dataSetsModulesArray.length; i++) { + dataSetsModule = dataSetsModulesArray[i] + dataSetModule.finalize() + } + filter = undefined + bot = undefined + thisObject = undefined + } + function isItADepenency(timeFrame, product) { let key = timeFrame + '-' + product diff --git a/DataSet.js b/DataSet.js index 66ded33de..066e6f7fc 100644 --- a/DataSet.js +++ b/DataSet.js @@ -8,6 +8,7 @@ exports.newDataSet = function newDataSet(BOT, logger) { node: undefined, networkNode: undefined, initialize: initialize, + finalize: finalize, getTextFile: getTextFile, createTextFile: createTextFile }; @@ -78,29 +79,36 @@ exports.newDataSet = function newDataSet(BOT, logger) { } /* Now we will see where do we need to fetch this data from. */ - let network = global.TASK_NETWORK.parentNode.parentNode.parentNode.parentNode.parentNode + let network = global.TASK_NETWORK let datasetProductDefinition = thisObject.node.parentNode - let datasetNetworkNode for (let i = 0; i < network.networkNodes.length; i++) { let networkNode = network.networkNodes[i] if (networkNode.dataStorage !== undefined) { if (networkNode.dataStorage.sessionIndependentData !== undefined) { - for (let j = 0; i < networkNode.dataStorage.sessionIndependentData.exchangeDataProducts.length; i++) { - let exchangeDataProduct = exchangeDataProducts[j] + for (let j = 0; j < networkNode.dataStorage.sessionIndependentData.exchangeDataProducts.length; j++) { + let exchangeDataProduct = networkNode.dataStorage.sessionIndependentData.exchangeDataProducts[j] for (let k = 0; k < exchangeDataProduct.singleMarketData.length; k++) { let singleMarketData = exchangeDataProduct.singleMarketData[k] - for (let m = 0; m < singleMarketData.dataProducts.length; m++) { - let dataProduct = singleMarketData.dataProducts[m] - if (dataProduct.referenceParent !== undefined) { - let productDefinition = dataProduct.referenceParent - if (datasetProductDefinition.id === productDefinition.id) { + if (singleMarketData.referenceParent !== undefined) { + let market = singleMarketData.referenceParent + let currentProcessMarket = bot.processNode.marketReference.referenceParent - /* We found where the data is located on the network. */ - thisObject.networkNode = networkNode - fileStorage = FILE_STORAGE.newFileStorage(logger, networkNode.code.host, networkNode.code.webPort); - callBackFunction(global.DEFAULT_OK_RESPONSE); - return + if (currentProcessMarket.id === market.id) { + for (let m = 0; m < singleMarketData.dataProducts.length; m++) { + let dataProduct = singleMarketData.dataProducts[m] + if (dataProduct.referenceParent !== undefined) { + let productDefinition = dataProduct.referenceParent + if (datasetProductDefinition.id === productDefinition.id) { + + /* We found where the data is located on the network. */ + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> Retrieving data from " + networkNode.name + " -> host = " + networkNode.code.host + ' -> port = ' + networkNode.code.webPort + '.'); } + + fileStorage = FILE_STORAGE.newFileStorage(logger, networkNode.code.host, networkNode.code.webPort); + callBackFunction(global.DEFAULT_OK_RESPONSE); + return + } + } } } } @@ -119,6 +127,13 @@ exports.newDataSet = function newDataSet(BOT, logger) { } } + function finalize() { + fileStorage = undefined + thisObject.networkNode = undefined + bot = undefined + thisObject = undefined + } + function getTextFile(pFolderPath, pFileName, callBackFunction) { try { diff --git a/EventServerClient.js b/EventServerClient.js index 3695225b2..a5de99f5a 100644 --- a/EventServerClient.js +++ b/EventServerClient.js @@ -1,4 +1,4 @@ -exports.newEventsServerClient = function newEventsServerClient() { +exports.newEventsServerClient = function newEventsServerClient(host, port) { const MODULE = "Task Server " const ERROR_LOG = true @@ -19,7 +19,14 @@ let WEB_SOCKETS_CLIENT const WEB_SOCKET = require('ws') - let port = process.env.WEB_SOCKETS_SERVER_PORT + + if (host === undefined) { + host = 'localhost' + } + if (port === undefined) { + port = process.env.WEB_SOCKETS_SERVER_PORT + } + return thisObject @@ -32,7 +39,7 @@ function setuptWebSockets(callBackFunction) { try { - WEB_SOCKETS_CLIENT = new WEB_SOCKET('ws://localhost:' + port ) + WEB_SOCKETS_CLIENT = new WEB_SOCKET('ws://' + host + ':' + port ) WEB_SOCKETS_CLIENT.onerror = error => { console.log('[ERROR] Task Server -> Event Server Client -> setuptWebSockets -> On connection error -> error = ' + JSON.stringify(error)) diff --git a/FileStorage.js b/FileStorage.js index 2778bc11c..7dff0ea37 100644 --- a/FileStorage.js +++ b/FileStorage.js @@ -52,7 +52,7 @@ exports.newFileStorage = function newFileStorage(logger, host, port) { logger.write(MODULE_NAME, '[INFO] FileStorage -> getTextFile -> fileLocation: ' + fileLocation) /* Here we actually reaad the file. */ - if (host === undefined) { + if (host === undefined || host === 'localhost') { /* We read the file from the local file system. */ const fs = require('fs') fs.readFile(fileLocation, onFileRead) @@ -124,7 +124,14 @@ exports.newFileStorage = function newFileStorage(logger, host, port) { if (canUsePrevious === true) { logger.write(MODULE_NAME, '[WARN] FileStorage -> getTextFile -> onFileRead -> Could read the file, but could not parse it as it is not a valid JSON. Will try to read the PREVIOUS version instead. -> file = ' + fileLocation) - fs.readFile(fileLocation + '.Previous.json', onPreviousFileRead) + if (host === undefined || host === 'localhost') { + /* We read the file from the local file system. */ + const fs = require('fs') + fs.readFile(fileLocation + '.Previous.json', onPreviousFileRead) + } else { + /* We read the file via a web server over http */ + getFileViaHTTP(fileLocation + '.Previous.json', onPreviousFileRead) + } function onPreviousFileRead(err, text) { @@ -452,7 +459,7 @@ exports.newFileStorage = function newFileStorage(logger, host, port) { function onEnd() { let fileContent = Buffer.concat(chunks).toString('utf8') - callback(global.DEFAULT_OK_RESPONSE, fileContent) + callback(null, fileContent) } } diff --git a/IndicatorBotProcessMainLoop.js b/IndicatorBotProcessMainLoop.js index bf61953d3..f74c6f1af 100644 --- a/IndicatorBotProcessMainLoop.js +++ b/IndicatorBotProcessMainLoop.js @@ -776,10 +776,13 @@ function onFinished(err) { try { - if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> startProcessFramework -> onFinished -> Entering function."); } processFramework.finalize() processFramework = undefined + dataDependencies.finalize() + dataDependencies = undefined + statusDependencies.finalize() + statusDependencies = undefined switch (err.result) { case global.DEFAULT_OK_RESPONSE.result: { @@ -876,11 +879,12 @@ processExecutionEvents.finish(onFinished); function onFinished(err) { - try { - if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Entering function."); } + processExecutionEvents.finalize() + processExecutionEvents = undefined + switch (err.result) { case global.DEFAULT_OK_RESPONSE.result: { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Execution finished well."); diff --git a/ProcessExecutionEvents.js b/ProcessExecutionEvents.js index 18ff8a70b..ba44bf7f2 100644 --- a/ProcessExecutionEvents.js +++ b/ProcessExecutionEvents.js @@ -8,27 +8,33 @@ */ const MODULE_NAME = "Process Execution Events"; + logger.fileName = MODULE_NAME let bot = BOT; let thisObject = { + networkNode: undefined, initialize, initialize, + finalize: finalize, start: start, finish: finish }; let processThisDependsOn; // This is the process this process needs to wait for it to finishes in order to start. - let currentProcessKey + let EVENT_SERVER_CLIENT = require('./EventServerClient.js'); + return thisObject; function initialize(callBackFunction) { try { + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> Entering function."); } let name = 'Not Depends on any Process' + logger.fileName = MODULE_NAME + "." + name; if (bot.processNode.referenceParent !== undefined) { @@ -48,9 +54,63 @@ } } } + + /* We need to find at which network node is running the process that we need to hear from when it finished. */ + let network = global.TASK_NETWORK + + for (let i = 0; i < network.networkNodes.length; i++) { + let networkNode = network.networkNodes[i] + if (networkNode.dataMining !== undefined) { + for (let j = 0; j < networkNode.dataMining.exchangeTasks.length; j++) { + let exchangeTasks = networkNode.dataMining.exchangeTasks[j] + for (let k = 0; k < exchangeTasks.taskManagers.length; k++) { + let taskManager = exchangeTasks.taskManagers[k] + for (let m = 0; m < taskManager.tasks.length; m++) { + let task = taskManager.tasks[m] + if (task.bot !== undefined) { + for (let n = 0; n < task.bot.processes.length; n++) { + let process = task.bot.processes[n] + if (process.marketReference !== undefined) { + if (process.marketReference.referenceParent !== undefined) { + let market = process.marketReference.referenceParent + let currentProcessMarket = bot.processNode.marketReference.referenceParent + + if (currentProcessMarket.id === market.id) { + if (process.referenceParent !== undefined) { + let processDefinition = process.referenceParent + if (processThisDependsOn.id === processDefinition.id) { + + /* We found where the task that runs the process definition we are waiting for is located on the network. */ + + thisObject.networkNode = networkNode + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> Connecting to Websockets Server " + networkNode.name + " -> host = " + networkNode.code.host + ' -> port = ' + networkNode.code.webSocketsPort + '.'); } + + EVENT_SERVER_CLIENT = EVENT_SERVER_CLIENT.newEventsServerClient(networkNode.code.host, networkNode.code.webSocketsPort) + EVENT_SERVER_CLIENT.initialize(onConnected) + + function onConnected() { + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> Connected to Websockets Server " + networkNode.name + " -> host = " + networkNode.code.host + ' -> port = ' + networkNode.code.webSocketsPort + '.'); } + + callBackFunction(global.DEFAULT_OK_RESPONSE); + return + } + return + } + } + } + } + } + } + } + } + } + } + } + } + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + logger.write(MODULE_NAME, "[ERROR] initialize -> Initialization Failed because we could not find where the task that runs the Process Definition we are waiting for is located within the network. Check the logs for more info."); + return } - - logger.fileName = MODULE_NAME + "." + name; callBackFunction(global.DEFAULT_OK_RESPONSE); } catch (err) { @@ -59,6 +119,16 @@ } } + function finalize() { + EVENT_SERVER_CLIENT.finalize() + EVENT_SERVER_CLIENT = undefined + processThisDependsOn = undefined + currentProcessKey = undefined + thisObject.networkNode = undefined + bot = undefined + thisObject = undefined + } + function start(callBackFunction) { try { @@ -81,9 +151,9 @@ if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> waitForDependantProcess -> key = " + key); } if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> waitForDependantProcess -> callerId = " + callerId); } - global.EVENT_SERVER_CLIENT.listenToEvent(key, 'Process Execution Finished', undefined, callerId, responseCallBack, eventsCallBack) + EVENT_SERVER_CLIENT.listenToEvent(key, 'Process Execution Finished', undefined, callerId, responseCallBack, eventsCallBack) - bot.processHeartBeat(undefined, undefined, "Waiting for " + processThisDependsOn.parentNode.parentNode.name + "->" + processThisDependsOn.parentNode.code.codeName + "->" + processThisDependsOn.code.codeName) + bot.processHeartBeat(undefined, undefined, "Waiting for " + thisObject.networkNode.name + "->" + processThisDependsOn.parentNode.parentNode.name + "->" + processThisDependsOn.parentNode.code.codeName + "->" + processThisDependsOn.code.codeName) function responseCallBack(message) { if (message.result !== global.DEFAULT_OK_RESPONSE.result) { @@ -102,7 +172,7 @@ /* We continue the normal flow after we learn the dependent process has finished its execution. */ if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> eventsCallBack -> stopListening to Process Execution Finished. "); } if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] start -> eventsCallBack -> subscriptionId = " + subscriptionId); } - global.EVENT_SERVER_CLIENT.stopListening(key, 'Process Execution Finished', subscriptionId) + EVENT_SERVER_CLIENT.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. */ diff --git a/SensorBotProcessMainLoop.js b/SensorBotProcessMainLoop.js index 5a5d55444..47c7a4361 100644 --- a/SensorBotProcessMainLoop.js +++ b/SensorBotProcessMainLoop.js @@ -584,11 +584,12 @@ processExecutionEvents.finish(onFinished); function onFinished(err) { - try { - if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Entering function."); } + processExecutionEvents.finalize() + processExecutionEvents = undefined + switch (err.result) { case global.DEFAULT_OK_RESPONSE.result: { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Execution finished well."); diff --git a/StatusDependencies.js b/StatusDependencies.js index 19e7485ca..8dcc1bd67 100644 --- a/StatusDependencies.js +++ b/StatusDependencies.js @@ -11,6 +11,7 @@ statusReports: new Map(), reportsByMainUtility: new Map(), initialize: initialize, + finalize: finalize, keys: [] }; @@ -165,4 +166,12 @@ } } + function finalize() { + statusReports.forEach(forEachStatusDependency) + function forEachStatusDependency(statusDependency) { + statusDependency.finalize() + } + bot = undefined + thisObject = undefined + } }; diff --git a/StatusReport.js b/StatusReport.js index 66839dac9..99fb2fc28 100644 --- a/StatusReport.js +++ b/StatusReport.js @@ -13,9 +13,11 @@ let bot = BOT; let thisObject = { + networkNode: undefined, mainUtility: undefined, file: undefined, // Here we have the JSON object representing the file content. initialize: initialize, + finalize: finalize, load: load, save: save, status: undefined @@ -30,7 +32,7 @@ /* Storage account to be used here. */ const FILE_STORAGE = require('./FileStorage.js'); - let fileStorage = FILE_STORAGE.newFileStorage(logger); + let fileStorage let sessionPath = '' @@ -93,17 +95,6 @@ statusDependencyNode.bottype = statusDependencyNode.referenceParent.parentNode.parentNode.type statusDependencyNode.dataMine = statusDependencyNode.referenceParent.parentNode.parentNode.parentNode.code.codeName - /* Let's see if the process is run monthly or not. */ - if (statusDependencyNode.referenceParent.parentNode.code.startMode !== undefined) { - if (statusDependencyNode.referenceParent.parentNode.code.startMode.allMonths !== undefined) { - if (statusDependencyNode.referenceParent.parentNode.code.startMode.oneMonth !== undefined) { - if (statusDependencyNode.referenceParent.parentNode.code.startMode.allMonths.run === "true" || statusDependencyNode.referenceParent.parentNode.code.startMode.oneMonth.run === "true") { - statusDependencyNode.processRunMonthly = true - } - } - } - } - /* We retrieve the report main utility */ if (statusDependencyNode.code !== undefined) { if (statusDependencyNode.code.mainUtility !== undefined) { @@ -122,7 +113,55 @@ sessionPath = bot.SESSION.folderName + "/" } - callBackFunction(global.DEFAULT_OK_RESPONSE); + /* Now we will see where do we need to fetch this status report from. */ + let network = global.TASK_NETWORK + let processThisDependsOn = statusDependencyNode.referenceParent.parentNode + + for (let i = 0; i < network.networkNodes.length; i++) { + let networkNode = network.networkNodes[i] + if (networkNode.dataMining !== undefined) { + for (let j = 0; j < networkNode.dataMining.exchangeTasks.length; j++) { + let exchangeTasks = networkNode.dataMining.exchangeTasks[j] + for (let k = 0; k < exchangeTasks.taskManagers.length; k++) { + let taskManager = exchangeTasks.taskManagers[k] + for (let m = 0; m < taskManager.tasks.length; m++) { + let task = taskManager.tasks[m] + if (task.bot !== undefined) { + for (let n = 0; n < task.bot.processes.length; n++) { + let process = task.bot.processes[n] + if (process.marketReference !== undefined) { + if (process.marketReference.referenceParent !== undefined) { + let market = process.marketReference.referenceParent + let currentProcessMarket = bot.processNode.marketReference.referenceParent + if (currentProcessMarket.id === market.id) { + if (process.referenceParent !== undefined) { + let processDefinition = process.referenceParent + if (processThisDependsOn.id === processDefinition.id) { + + /* We found where the task that runs the process definition this status report depends on and where it is located on the network. */ + + thisObject.networkNode = networkNode + if (global.LOG_CONTROL[MODULE_NAME].logInfo === true) { logger.write(MODULE_NAME, "[INFO] initialize -> Retrieving status report from " + networkNode.name + " -> host = " + networkNode.code.host + ' -> port = ' + networkNode.code.webPort + '.'); } + + fileStorage = FILE_STORAGE.newFileStorage(logger, networkNode.code.host, networkNode.code.webPort); + callBackFunction(global.DEFAULT_OK_RESPONSE); + return + } + } + } + } + } + } + } + } + } + } + } + } + + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + logger.write(MODULE_NAME, "[ERROR] initialize -> Initialization Failed because we could not find where the data of this status report is located within the network. Check the logs for more info."); + } catch (err) { logger.write(MODULE_NAME, "[ERROR] initialize -> err = "+ err.stack); @@ -130,6 +169,13 @@ } } + function finalize() { + fileStorage = undefined + thisObject.networkNode = undefined + bot = undefined + thisObject = undefined + } + function load(callBackFunction) { try { diff --git a/TradingBotProcessMainLoop.js b/TradingBotProcessMainLoop.js index 461b648a8..463d16fee 100644 --- a/TradingBotProcessMainLoop.js +++ b/TradingBotProcessMainLoop.js @@ -694,6 +694,10 @@ if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> startProcessFramework -> onFinished -> Entering function."); } processFramework.finalize() processFramework = undefined + dataDependencies.finalize() + dataDependencies = undefined + statusDependencies.finalize() + statusDependencies = undefined switch (err.result) { case global.DEFAULT_OK_RESPONSE.result: { @@ -792,8 +796,9 @@ processExecutionEvents.finish(onFinished); function onFinished(err) { - try { + processExecutionEvents.finalize() + processExecutionEvents = undefined if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> finishProcessExecutionEvents -> onFinished -> Entering function."); } diff --git a/server.js b/server.js index 310145b83..f32a3bc13 100644 --- a/server.js +++ b/server.js @@ -131,7 +131,7 @@ function preLoader() { } } catch (err) { console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE -> ' + err.stack) - console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE = ' + JSON.stringify(global.TASK_NODE)) + console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE = ' + JSON.stringify(global.TASK_NODE).substring(0, 1000)) } } else { /* This process was started not by the Task Manager, but independently (most likely for debugging purposes). In this case we listen to an event with the Task Info that should be emitted at the UI */ @@ -146,7 +146,7 @@ function preLoader() { } } catch (err) { console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE -> ' + err.stack) - console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE = ' + JSON.stringify(global.TASK_NODE)) + console.log('[ERROR] Task Server -> server -> preLoader -> global.TASK_NODE = ' + JSON.stringify(global.TASK_NODE).substring(0, 1000)) } } }