diff --git a/src/DailyFiles.js b/src/DailyFiles.js index fe6eefe52..8e58e1e52 100644 --- a/src/DailyFiles.js +++ b/src/DailyFiles.js @@ -6,6 +6,7 @@ function newDailyFiles () { logger.fileName = MODULE_NAME let thisObject = { + eventHandler: undefined, getFileCursor: getFileCursor, setDatetime: setDatetime, setTimePeriod: setTimePeriod, @@ -17,17 +18,19 @@ function newDailyFiles () { let filesLoaded = 0 let expectedFiles = 0 - let fileCloud - let fileCursors = new Map() - let callBackWhenFileReceived + thisObject.eventHandler = newEventHandler() + return thisObject function finalize () { try { + thisObject.eventHandler.finalize() + thisObject.eventHandler = undefined + filesLoaded = undefined expectedFiles = undefined fileCloud = undefined @@ -79,9 +82,9 @@ function newDailyFiles () { case GLOBAL.CUSTOM_FAIL_RESPONSE.result: { if (err.message === 'File does not exist.') { - err.message = 'Dataset Unavailable.' - callBackFunction(err) - return + beginDateRange = new Date() + endDateRange = new Date() + break } if (err.message === 'Missing Configuration.') { if (ERROR_LOG === true) { logger.write('[WARN] initialize -> onFileReceived -> The needed configuration for the dateRange at the dataSet of the product was not found.') } @@ -106,7 +109,8 @@ function newDailyFiles () { if (pSet.validPeriods.includes(periodName) === true) { let fileCursor = newFileCursor() - fileCursor.initialize(fileCloud, pDevTeam, pBot, pSet, exchange, pMarket, periodName, periodTime, pDatetime, pTimePeriod, beginDateRange, endDateRange, onInitialized) + fileCursor.eventHandler = thisObject.eventHandler // We share our event handler with each file cursor, so that they can raise events there when files are changed.s + fileCursor.initialize(fileCloud, pDevTeam, pBot, pProduct, pSet, exchange, pMarket, periodName, periodTime, pDatetime, pTimePeriod, beginDateRange, endDateRange, onInitialized) function onInitialized (err) { try { switch (err.result) { diff --git a/src/FileCursor.js b/src/FileCursor.js index 6b9f0c487..e856afedd 100644 --- a/src/FileCursor.js +++ b/src/FileCursor.js @@ -9,6 +9,7 @@ function newFileCursor () { let cursorDate let thisObject = { + eventHandler: undefined, // received from parent reload: reload, setDatetime: setDatetime, setTimePeriod: setTimePeriod, @@ -26,22 +27,35 @@ function newFileCursor () { let fileCloud let devTeam let bot + let product let thisSet let periodName let timePeriod let beginDateRange let endDateRange - let intervalHandle let finalized = false + let eventSubscriptionIdDatasetUpdated + return thisObject function finalize () { try { - if (INFO_LOG === true) { logger.write('[INFO] finalize -> Entering function.') } + systemEventHandler.stopListening('Dataset Updated', eventSubscriptionIdDatasetUpdated) + thisObject.eventHandler = undefined - clearInterval(intervalHandle) + market = undefined + exchange = undefined + fileCloud = undefined + devTeam = undefined + bot = undefined + product = undefined + thisSet = undefined + periodName = undefined + timePeriod = undefined + beginDateRange = undefined + endDateRange = undefined thisObject.files = undefined cursorDate = undefined @@ -54,16 +68,14 @@ function newFileCursor () { } } - function initialize (pFileCloud, pDevTeam, pBot, pSet, pExchange, pMarket, pPeriodName, pTimePeriod, pCursorDate, pCurrentTimePeriod, pBeginDateRange, pEndDateRange, callBackFunction) { + function initialize (pFileCloud, pDevTeam, pBot, pProduct, pSet, pExchange, pMarket, pPeriodName, pTimePeriod, pCursorDate, pCurrentTimePeriod, pBeginDateRange, pEndDateRange, callBackFunction) { try { - if (INFO_LOG === true) { logger.write('[INFO] initialize -> Entering function.') } - if (INFO_LOG === true) { logger.write('[INFO] initialize -> key = ' + pDevTeam.codeName + '-' + pBot.codeName + '-' + pSet.codeName + '-' + pPeriodName) } - market = pMarket exchange = pExchange fileCloud = pFileCloud devTeam = pDevTeam bot = pBot + product = pProduct thisSet = pSet periodName = pPeriodName cursorDate = removeTime(pCursorDate) @@ -71,95 +83,42 @@ function newFileCursor () { beginDateRange = pBeginDateRange endDateRange = pEndDateRange - if (INFO_LOG === true) { logger.write('[INFO] initialize -> market = ' + market) } - if (INFO_LOG === true) { logger.write('[INFO] initialize -> periodName = ' + periodName) } - if (INFO_LOG === true) { logger.write('[INFO] initialize -> cursorDate = ' + cursorDate) } - if (INFO_LOG === true) { logger.write('[INFO] initialize -> beginDateRange = ' + beginDateRange) } - if (INFO_LOG === true) { logger.write('[INFO] initialize -> endDateRange = ' + endDateRange) } - - setTimePeriod(pCurrentTimePeriod, pCursorDate) - - intervalHandle = setInterval(updateFiles, timePeriod) + let key = devTeam.codeName + '-' + bot.codeName + '-' + product.codeName + '-' + thisSet.codeName + systemEventHandler.listenToEvent(key, 'Dataset Updated', undefined, key + '-' + periodName, onResponse, updateFiles) callBackFunction(GLOBAL.DEFAULT_OK_RESPONSE) + + function onResponse (message) { + eventSubscriptionIdDatasetUpdated = message.eventSubscriptionId + } } catch (err) { if (ERROR_LOG === true) { logger.write('[ERROR] initialize -> err = ' + err.stack) } callBackFunction(GLOBAL.DEFAULT_FAIL_RESPONSE) } } - function updateFiles () { + function updateFiles (message) { try { if (finalized === true) { return } - if (INFO_LOG === true) { logger.write('[INFO] updateFiles -> Entering function.') } - /* + if (message.event === undefined || message.event.lastFile === undefined) { return } - In order to know if we need to update any file we follow this logic: - - 1. If the current date is already a file in the cursor, we need to update it. - 2. If the current date is one day ahead of a file in the cursor, we need to add this one to the cursor. - - */ - - let targetDate + let targetDate = new Date(message.event.lastFile) let dateString let file - /* Situation 1 */ - - targetDate = new Date() dateString = targetDate.getUTCFullYear() + '-' + pad(targetDate.getUTCMonth() + 1, 2) + '-' + pad(targetDate.getUTCDate(), 2) - file = thisObject.files.get(dateString) - - if (file !== undefined) { - fileCloud.getFile(devTeam, bot, thisSet, exchange, market, periodName, targetDate, undefined, undefined, onFileReceived) - - return - } - - /* Situation 2 */ - - targetDate = new Date((new Date()).valueOf() - ONE_DAY_IN_MILISECONDS) - dateString = targetDate.getUTCFullYear() + '-' + pad(targetDate.getUTCMonth() + 1, 2) + '-' + pad(targetDate.getUTCDate(), 2) - - file = thisObject.files.get(dateString) // This is from yesterday - - targetDate = new Date() - dateString = targetDate.getUTCFullYear() + '-' + pad(targetDate.getUTCMonth() + 1, 2) + '-' + pad(targetDate.getUTCDate(), 2) - - if (file !== undefined) { - fileCloud.getFile(devTeam, bot, thisSet, exchange, market, periodName, targetDate, undefined, undefined, onFileReceived) - - return - } + fileCloud.getFile(devTeam, bot, thisSet, exchange, market, periodName, targetDate, undefined, undefined, onFileReceived) function onFileReceived (err, file) { try { if (finalized === true) { return } - switch (err.result) { - case GLOBAL.DEFAULT_OK_RESPONSE.result: { - break - } - - case GLOBAL.DEFAULT_FAIL_RESPONSE.result: { - return - } - - default: { - return - } + if (err.result === GLOBAL.DEFAULT_OK_RESPONSE.result) { + thisObject.files.set(dateString, file) + thisObject.eventHandler.raiseEvent('Files Updated') } - - thisObject.files.set(dateString, file) - - if (INFO_LOG === true) { logger.write('[INFO] updateFiles -> onFileReceived -> File updated.') } - if (INFO_LOG === true) { logger.write('[INFO] updateFiles -> onFileReceived -> devTeam = ' + devTeam.codeName) } - if (INFO_LOG === true) { logger.write('[INFO] updateFiles -> onFileReceived -> bot = ' + bot.codeName) } - if (INFO_LOG === true) { logger.write('[INFO] updateFiles -> onFileReceived -> thisSet = ' + thisSet.codeName) } - if (INFO_LOG === true) { logger.write('[INFO] updateFiles -> onFileReceived -> dateString = ' + dateString) } } catch (err) { if (ERROR_LOG === true) { logger.write('[ERROR] updateFiles -> onFileReceived -> err = ' + err.stack) } callBackFunction(GLOBAL.DEFAULT_FAIL_RESPONSE) @@ -202,23 +161,13 @@ function newFileCursor () { } if (Math.abs(positionB - positionA) <= 1) { - if (INFO_LOG === true) { logger.write('[INFO] setTimePeriod -> Will EXIT Saving Mode.') } - if (INFO_LOG === true) { logger.write('[INFO] setTimePeriod -> pTimePeriod = ' + convertTimePeriodToName(pTimePeriod)) } - if (INFO_LOG === true) { logger.write('[INFO] setTimePeriod -> timePeriod = ' + convertTimePeriodToName(timePeriod)) } - exitSavingMode() } else { - if (INFO_LOG === true) { logger.write('[INFO] setTimePeriod -> Will ENTER Saving Mode.') } - if (INFO_LOG === true) { logger.write('[INFO] setTimePeriod -> pTimePeriod = ' + convertTimePeriodToName(pTimePeriod)) } - if (INFO_LOG === true) { logger.write('[INFO] setTimePeriod -> timePeriod = ' + convertTimePeriodToName(timePeriod)) } - enterSavingMode() } function enterSavingMode () { try { - if (INFO_LOG === true) { logger.write('[INFO] setTimePeriod -> enterSavingMode -> Entering function.') } - switch (timePeriod) { case _45_MINUTES_IN_MILISECONDS: @@ -296,8 +245,6 @@ function newFileCursor () { function exitSavingMode () { try { - if (INFO_LOG === true) { logger.write('[INFO] setTimePeriod -> exitSavingMode -> Entering function.') } - switch (timePeriod) { case _45_MINUTES_IN_MILISECONDS: @@ -380,7 +327,6 @@ function newFileCursor () { function setDatetime (pDatetime) { try { if (finalized === true) { return } - if (INFO_LOG === true) { logger.write('[INFO] setDatetime -> Entering function.') } if (pDatetime === undefined) { if (ERROR_LOG === true) { logger.write('[ERROR] setDatetime -> Received undefined datetime.') } @@ -396,7 +342,6 @@ function newFileCursor () { function reload (callBackFunction) { try { if (finalized === true) { return } - if (INFO_LOG === true) { logger.write('[INFO] reload -> Entering function.') } getFiles(callBackFunction) @@ -410,7 +355,6 @@ function newFileCursor () { function getFiles (callBackFunction) { try { if (finalized === true) { return } - if (INFO_LOG === true) { logger.write('[INFO] getFiles -> Entering function.') } let i = 0 let j = 0 @@ -421,8 +365,6 @@ function newFileCursor () { function getNextFile () { try { - if (INFO_LOG === true) { logger.write('[INFO] getFiles -> getNextFile -> Entering function.') } - let targetDate = new Date(cursorDate) targetDate.setUTCDate(targetDate.getUTCDate() + j) @@ -550,7 +492,6 @@ function newFileCursor () { function collectGarbage (callBackFunction) { try { if (finalized === true) { return } - if (INFO_LOG === true) { logger.write('[INFO] collectGarbage -> Entering function.') } date = removeTime(cursorDate) @@ -577,4 +518,3 @@ function newFileCursor () { return minCursorSize } } - diff --git a/src/FileSequence.js b/src/FileSequence.js index 24ebd4033..cfca4992d 100644 --- a/src/FileSequence.js +++ b/src/FileSequence.js @@ -29,18 +29,27 @@ function newFileSequence () { thisObject.eventHandler = newEventHandler() - let intervalHandle + let eventSubscriptionIdDatasetUpdated return thisObject function finalize () { try { - clearInterval(intervalHandle) + systemEventHandler.stopListening('Dataset Updated', eventSubscriptionIdDatasetUpdated) + + thisObject.eventHandler.finalize() + thisObject.eventHandler = undefined filesLoaded = undefined fileCloud = undefined files = undefined maxSequence = undefined + market = undefined + devTeam = undefined + bot = undefined + thisSet = undefined + product = undefined + finalized = true } catch (err) { if (ERROR_LOG === true) { logger.write('[ERROR] finalize -> err = ' + err.stack) } @@ -55,8 +64,6 @@ function newFileSequence () { throw 'Exchange not supoorted by this pProduct of the ecosystem! - pDevTeam.codeName = ' + pDevTeam.codeName + ', pBot.codeName = ' + pBot.codeName + ', pProduct.codeName = ' + pProduct.codeName + ', pExchange = ' + pExchange } - intervalHandle = setInterval(updateFiles, _1_MINUTE_IN_MILISECONDS) - market = pMarket devTeam = pDevTeam bot = pBot @@ -66,12 +73,20 @@ function newFileSequence () { fileCloud = newFileCloud() fileCloud.initialize(bot) + let key = devTeam.codeName + '-' + bot.codeName + '-' + product.codeName + '-' + thisSet.codeName + systemEventHandler.listenToEvent(key, 'Dataset Updated', undefined, key, onResponse, updateFiles) + + function onResponse (message) { + eventSubscriptionIdDatasetUpdated = message.eventSubscriptionId + } + /* First we will get the sequence max number */ fileCloud.getFile(devTeam, bot, thisSet, exchange, market, undefined, undefined, 'Sequence', undefined, onSequenceFileReceived) function onSequenceFileReceived (err, file) { try { + if (finalized === true) { return } initialized = true switch (err.result) { @@ -113,6 +128,7 @@ function newFileSequence () { function onFileReceived (err, file) { try { + if (finalized === true) { return } switch (err.result) { case GLOBAL.DEFAULT_OK_RESPONSE.result: { break @@ -179,6 +195,7 @@ function newFileSequence () { function onSequenceFileReceived (err, file) { try { + if (finalized === true) { return } switch (err.result) { case GLOBAL.DEFAULT_OK_RESPONSE.result: { break @@ -213,6 +230,7 @@ function newFileSequence () { function onFileReceived (err, file) { try { + if (finalized === true) { return } switch (err.result) { case GLOBAL.DEFAULT_OK_RESPONSE.result: { break diff --git a/src/MarketFiles.js b/src/MarketFiles.js index a682c491c..05aa887ba 100644 --- a/src/MarketFiles.js +++ b/src/MarketFiles.js @@ -27,6 +27,7 @@ function newMarketFiles () { let devTeam let bot let thisSet + let product let periodName thisObject.eventHandler = newEventHandler() @@ -34,14 +35,28 @@ function newMarketFiles () { let intervalHandle let finalized = false + let eventSubscriptionIdDatasetUpdated + return thisObject function finalize () { try { - clearInterval(intervalHandle) + systemEventHandler.stopListening('Dataset Updated', eventSubscriptionIdDatasetUpdated) + + thisObject.eventHandler.finalize() + thisObject.eventHandler = undefined filesLoaded = undefined files = undefined + + market = undefined + exchange = undefined + devTeam = undefined + bot = undefined + thisSet = undefined + product = undefined + periodName = undefined + finalized = true } catch (err) { if (ERROR_LOG === true) { logger.write('[ERROR] finalize -> err = ' + err.stack) } @@ -56,12 +71,11 @@ function newMarketFiles () { throw 'Exchange not supoorted by this pProduct of the ecosystem! - pDevTeam.codeName = ' + pDevTeam.codeName + ', pBot.codeName = ' + pBot.codeName + ', pProduct.codeName = ' + pProduct.codeName + ', pExchange = ' + pExchange } - intervalHandle = setInterval(updateFiles, _10_MINUTES_IN_MILISECONDS) - market = pMarket devTeam = pDevTeam bot = pBot thisSet = pSet + product = pProduct fileCloud = newFileCloud() fileCloud.initialize(pBot) @@ -77,6 +91,7 @@ function newMarketFiles () { function onFileReceived (err, file) { try { + if (finalized === true) { return } if (err.result === GLOBAL.DEFAULT_OK_RESPONSE.result) { files.set(periodTime, file) filesLoaded++ @@ -85,7 +100,14 @@ function newMarketFiles () { } if (filesLoaded + filesNotLoaded === marketFilesPeriods.length) { + let key = devTeam.codeName + '-' + bot.codeName + '-' + product.codeName + '-' + thisSet.codeName + systemEventHandler.listenToEvent(key, 'Dataset Updated', undefined, key + '-' + periodName, onResponse, updateFiles) + callBackFunction(GLOBAL.DEFAULT_OK_RESPONSE, thisObject) + + function onResponse (message) { + eventSubscriptionIdDatasetUpdated = message.eventSubscriptionId + } } } catch (err) { if (ERROR_LOG === true) { logger.write('[ERROR] initialize -> onFileReceived -> err = ' + err.stack) } @@ -116,11 +138,12 @@ function newMarketFiles () { function onFileReceived (err, file) { try { + if (finalized === true) { return } files.set(periodTime, file) updatedFiles++ if (updatedFiles === marketFilesPeriods.length) { - thisObject.eventHandler.raiseEvent('Files Updated', undefined) + thisObject.eventHandler.raiseEvent('Files Updated') } } catch (err) { if (ERROR_LOG === true) { logger.write('[ERROR] updateFiles -> onFileReceived -> err = ' + err.stack) }