fetching data from networknodes
This commit is contained in:
1 parent
ea519e2525
commit
480ab1b78e
3 files changed
+90
-12
No files matched your search
+37
-2
@@ -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);
|
||||
|
||||
+49
-8
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
Reference in new issue
Block a user