diff --git a/DataSet.js b/DataSet.js index 434a86ce1..66ded33de 100644 --- a/DataSet.js +++ b/DataSet.js @@ -6,6 +6,7 @@ exports.newDataSet = function newDataSet(BOT, logger) { let thisObject = { node: undefined, + networkNode: undefined, initialize: initialize, getTextFile: getTextFile, createTextFile: createTextFile @@ -14,7 +15,7 @@ exports.newDataSet = function newDataSet(BOT, logger) { /* Storage account to be used here. */ const FILE_STORAGE = require('./FileStorage.js'); - let fileStorage = FILE_STORAGE.newFileStorage(logger); + let fileStorage return thisObject; @@ -76,7 +77,41 @@ exports.newDataSet = function newDataSet(BOT, logger) { return } - callBackFunction(global.DEFAULT_OK_RESPONSE); + /* Now we will see where do we need to fetch this data from. */ + let network = global.TASK_NETWORK.parentNode.parentNode.parentNode.parentNode.parentNode + 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 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) { + + /* 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 + } + } + } + } + } + } + } + } + + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + logger.write(MODULE_NAME, "[ERROR] initialize -> Initialization Failed because we could not find where the data of this dataset is located within the network. Check the logs for more info."); } catch (err) { logger.write(MODULE_NAME, "[ERROR] initialize -> err = "+ err.stack); diff --git a/FileStorage.js b/FileStorage.js index 625c78fd5..2778bc11c 100644 --- a/FileStorage.js +++ b/FileStorage.js @@ -1,7 +1,7 @@ const path = require('path') -exports.newFileStorage = function newFileStorage(logger) { +exports.newFileStorage = function newFileStorage(logger, host, port) { const MODULE_NAME = 'FileStorage' const MAX_RETRY = 30 @@ -51,9 +51,15 @@ exports.newFileStorage = function newFileStorage(logger) { logger.write(MODULE_NAME, '[INFO] FileStorage -> getTextFile -> fileLocation: ' + fileLocation) - /* Here we actually write the file. */ - const fs = require('fs') - fs.readFile(fileLocation, onFileRead) + /* Here we actually reaad the file. */ + if (host === undefined) { + /* We read the file from the local file system. */ + const fs = require('fs') + fs.readFile(fileLocation, onFileRead) + } else { + /* We read the file via a web server over http */ + getFileViaHTTP(fileLocation, onFileRead) + } function onFileRead(err, text) { let retryTimeToUse = FAST_RETRY_TIME_IN_MILISECONDS @@ -390,9 +396,8 @@ exports.newFileStorage = function newFileStorage(logger) { } } - /* Function to create folders of missing folders at any path. */ - - function mkDirByPathSync(targetDir, { isRelativeToScript = false } = {}) { + /* Function to create folders of missing folders at any path. */ + function mkDirByPathSync(targetDir, { isRelativeToScript = false } = {}) { const sep = '/'; const initDir = path.isAbsolute(targetDir) ? sep : ''; const baseDir = isRelativeToScript ? __dirname : '.'; @@ -420,5 +425,41 @@ exports.newFileStorage = function newFileStorage(logger) { return curDir; }, initDir); - } + } + + function getFileViaHTTP(filePath, callback) { + try { + + let http = require('http'); + let url = 'http://' + host + + ':' + port + + '/Storage/' + + filePath + + http.get(url, onResponse); + + function onResponse(response) { + + const chunks = [] + + response.on('data', onMessegesArrived) + response.on('end', onEnd) + + function onMessegesArrived(chunk) { + chunks.push(chunk) + } + + function onEnd() { + let fileContent = Buffer.concat(chunks).toString('utf8') + + callback(global.DEFAULT_OK_RESPONSE, fileContent) + } + } + + } catch (err) { + logger.write(MODULE_NAME, "[ERROR] getFileViaHTTP -> err = " + err.stack); + callBackFunction(global.DEFAULT_FAIL_RESPONSE); + abort = true + } + } } diff --git a/server.js b/server.js index 631ed3edb..310145b83 100644 --- a/server.js +++ b/server.js @@ -125,7 +125,8 @@ function preLoader() { global.EVENT_SERVER_CLIENT.raiseEvent('Task Manager - ' + taskId, 'Nodejs Process Ready for Task') function eventReceived(message) { global.TASK_NODE = message - global.TASK_NODE = JSON.parse(message.event.definition) + global.TASK_NODE = JSON.parse(message.event.taskDefinition) + global.TASK_NETWORK = JSON.parse(message.event.networkDefinition) bootLoader() } } catch (err) { @@ -139,7 +140,8 @@ function preLoader() { global.EVENT_SERVER_CLIENT.listenToEvent('Task Server', 'Debug Task Started', undefined, 'Task Server', undefined, startDebugging) function startDebugging(message) { global.TASK_NODE = message - global.TASK_NODE = JSON.parse(message.event.definition) + global.TASK_NODE = JSON.parse(message.event.taskDefinition) + global.TASK_NETWORK = JSON.parse(message.event.networkDefinition) bootLoader() } } catch (err) {