From ec4450f0fee9834ea2572bd77ee829f334c33fb3 Mon Sep 17 00:00:00 2001 From: Luis Fernando Molina Date: Fri, 25 Sep 2020 20:24:32 +0200 Subject: [PATCH] Session and Task Syncronization with the backend --- BackendServers/taskManagerServer.js | 9 ++-- .../FunctionLibraries/SessionFunctions.js | 35 +++++++++++-- .../FunctionLibraries/TaskFunctions.js | 6 +-- .../FunctionLibraries/UiObjectsFromNodes.js | 50 ++++++++----------- CanvasApp/DesignSpace/Workspace/Workspace.js | 32 +++++------- CanvasApp/FloatingSpace/UiObject.js | 3 +- .../LowFrequencyTrading/TradingProcess.js | 4 +- TaskServer/Session.js | 29 ++++++++--- TaskServer/TradingBot.js | 10 ++-- TaskServer/server.js | 12 +++++ 10 files changed, 115 insertions(+), 75 deletions(-) diff --git a/BackendServers/taskManagerServer.js b/BackendServers/taskManagerServer.js index 381c4237a..7ab879449 100644 --- a/BackendServers/taskManagerServer.js +++ b/BackendServers/taskManagerServer.js @@ -41,7 +41,7 @@ eventsServerClient.listenToEvent('Task Manager', 'Task Status', undefined, undefined, undefined, taskStatus) console.log('Task Manager Server Started.') - + function runTask(message) { //console.log('[INFO] BackendServers -> Task Manager Server -> runTask -> Entering function.') @@ -165,14 +165,13 @@ //console.log('[INFO] BackendServers -> Task Manager Server -> taskStatus -> Task Id = ' + message.event.taskId) let task = tasksMap.get(message.event.taskId) - - let event = {status: 'Undefined'} + let event = {} if (task) { event.status = 'Task Process Running' - eventsServerClient.raiseEvent('Task Client - ' + task.id, 'Task Status', event.status) + eventsServerClient.raiseEvent('Task Client - ' + message.event.taskId, 'Task Status', event) } else { event.status = 'Task Process Not Running' - eventsServerClient.raiseEvent('Task Client - ' + task.id, 'Task Status', event.status) + eventsServerClient.raiseEvent('Task Client - ' + message.event.taskId, 'Task Status', event) } } } diff --git a/CanvasApp/DesignSpace/Workspace/FunctionLibraries/SessionFunctions.js b/CanvasApp/DesignSpace/Workspace/FunctionLibraries/SessionFunctions.js index 7a92aa824..6a53cd293 100644 --- a/CanvasApp/DesignSpace/Workspace/FunctionLibraries/SessionFunctions.js +++ b/CanvasApp/DesignSpace/Workspace/FunctionLibraries/SessionFunctions.js @@ -1,11 +1,41 @@ function newSessionFunctions() { thisObject = { + syncronizeSessionWithBackEnd: syncronizeSessionWithBackEnd, runSession: runSession, stopSession: stopSession } return thisObject + function syncronizeSessionWithBackEnd(node) { + let networkNode = validations(node) + if (networkNode === undefined) { + /* Nodes that do not belong to a network can not get ready. */ + return + } + + let eventsServerClient = canvas.designSpace.workspace.eventsServerClients.get(networkNode.id) + + /* First we setup everything so as to listen to the response from the Task Server */ + let eventSubscriptionIdOnStatus + let key = node.name + '-' + node.type + '-' + node.id + eventsServerClient.listenToEvent(key, 'Status Response', undefined, node.id, onResponse, onStatus) + + function onResponse(message) { + eventSubscriptionIdOnStatus = message.eventSubscriptionId + } + + function onStatus(message) { + eventsServerClient.stopListening(key, eventSubscriptionIdOnStatus, node.id) + if (message.event.status === 'Session Runnning' ) { + node.payload.uiObject.menu.internalClick('Run Session') + } + } + + /* Second we ask the Task Server if this Session is Running. */ + eventsServerClient.raiseEvent(key, 'Session Status') + } + function runSession(node, functionLibraryProtocolNode, functionLibraryDependenciesFilter, resume, callBackFunction) { let networkNode = validations(node) if (networkNode === undefined) { @@ -179,11 +209,6 @@ function newSessionFunctions() { let networkNode = taskManager.payload.parentNode.payload.parentNode.payload.parentNode.payload.parentNode.payload.parentNode - if (node.payload.parentNode.payload.uiObject.isRunning !== true) { - node.payload.uiObject.setErrorMessage('Session needs a Process Instance parent to be running.') - return - } - if (node.tradingSystemReference === undefined) { node.payload.uiObject.setErrorMessage('Session needs a child Trading System Reference.') return diff --git a/CanvasApp/DesignSpace/Workspace/FunctionLibraries/TaskFunctions.js b/CanvasApp/DesignSpace/Workspace/FunctionLibraries/TaskFunctions.js index a7913454e..615c8dd2a 100644 --- a/CanvasApp/DesignSpace/Workspace/FunctionLibraries/TaskFunctions.js +++ b/CanvasApp/DesignSpace/Workspace/FunctionLibraries/TaskFunctions.js @@ -1,6 +1,6 @@ function newTaskFunctions() { thisObject = { - getReady: getReady, + syncronizeTaskWithBackEnd: syncronizeTaskWithBackEnd, runTask: runTask, stopTask: stopTask, @@ -35,7 +35,7 @@ function newTaskFunctions() { return thisObject - function getReady(node) { + function syncronizeTaskWithBackEnd(node) { let networkNode = validations(node) if (networkNode === undefined) { /* Nodes that do not belong to a network can not get ready. */ @@ -56,7 +56,7 @@ function newTaskFunctions() { function onStatus(message) { eventsServerClient.stopListening(key, eventSubscriptionIdOnStatus, node.id) if (message.event.status === 'Task Process Running' ) { - node.payload.uiObject.run() + node.payload.uiObject.menu.internalClick('Run Task') } } diff --git a/CanvasApp/DesignSpace/Workspace/FunctionLibraries/UiObjectsFromNodes.js b/CanvasApp/DesignSpace/Workspace/FunctionLibraries/UiObjectsFromNodes.js index c3f52c5a9..19c135c59 100644 --- a/CanvasApp/DesignSpace/Workspace/FunctionLibraries/UiObjectsFromNodes.js +++ b/CanvasApp/DesignSpace/Workspace/FunctionLibraries/UiObjectsFromNodes.js @@ -1,7 +1,7 @@ function newUiObjectsFromNodes() { thisObject = { - runTasks: runTasks, - runSessions: runSessions, + syncronizeTasksFoundAtWorkspaceWithBackEnd: syncronizeTasksFoundAtWorkspaceWithBackEnd, + syncronizeSessionsFoundAtWorkspaceWithBackEnd: syncronizeSessionsFoundAtWorkspaceWithBackEnd, playTutorials: playTutorials, recreateWorkspace: recreateWorkspace, getNodeById: getNodeById, @@ -13,8 +13,8 @@ function newUiObjectsFromNodes() { let mapOfReferenceChildren = new Map() let mapOfNodes - let tasksToRun - let sessionsToRun + let tasksFoundAtWorkspace + let sessionsFoundAtWorkspace let tutorialsToPlay return thisObject @@ -25,8 +25,8 @@ function newUiObjectsFromNodes() { function recreateWorkspace(node, callBackFunction) { mapOfNodes = new Map() - tasksToRun = [] - sessionsToRun = [] + tasksFoundAtWorkspace = [] + sessionsFoundAtWorkspace = [] tutorialsToPlay = [] removeNullRootNodes() @@ -150,20 +150,20 @@ function newUiObjectsFromNodes() { } } - function runTasks() { - for (let i = 0; i < tasksToRun.length; i++) { - let node = tasksToRun[i] - node.payload.uiObject.menu.internalClick('Run Task') + function syncronizeTasksFoundAtWorkspaceWithBackEnd(functionLibraryTaskFunctions) { + for (let i = 0; i < tasksFoundAtWorkspace.length; i++) { + let node = tasksFoundAtWorkspace[i] + functionLibraryTaskFunctions.syncronizeTaskWithBackEnd(node) } - tasksToRun = undefined + tasksFoundAtWorkspace = undefined } - function runSessions() { - for (let i = 0; i < sessionsToRun.length; i++) { - let node = sessionsToRun[i] - node.payload.uiObject.menu.internalClick('Run Session') + function syncronizeSessionsFoundAtWorkspaceWithBackEnd(functionLibrarySessionFunctions) { + for (let i = 0; i < sessionsFoundAtWorkspace.length; i++) { + let node = sessionsFoundAtWorkspace[i] + functionLibrarySessionFunctions.syncronizeSessionWithBackEnd(node) } - sessionsToRun = undefined + sessionsFoundAtWorkspace = undefined } function playTutorials() { @@ -625,26 +625,18 @@ function newUiObjectsFromNodes() { /* This is the point where we build a map with all nodes present at the workspace */ mapOfNodes.set(node.id, node) - /* Check if there are tasks to run */ + /* We will collect all tasks at the workspace in order to later syncronize them with the backend */ if (userAddingNew === false && uiObjectType === 'Task' && node.savedPayload !== undefined) { - if (node.savedPayload.uiObject.isRunning === true) { - if (node.payload.parentNode !== undefined) { - if (tasksToRun !== undefined) { // it might be undefined when you are spawning a task that was running while backed up - tasksToRun.push(node) - } - } + if (tasksFoundAtWorkspace !== undefined) { // it might be undefined when you are spawning a task that was backed up + tasksFoundAtWorkspace.push(node) } } /* Check if there are sessions to run */ if (userAddingNew === false && node.savedPayload !== undefined) { if (uiObjectType === 'Live Trading Session' || uiObjectType === 'Forward Testing Session' || uiObjectType === 'Backtesting Session' || uiObjectType === 'Paper Trading Session') { - if (node.savedPayload.uiObject.isRunning === true) { - if (node.payload.parentNode !== undefined) { - if (sessionsToRun !== undefined) { // it might be undefined when you are spawning a session that was running while backed up - sessionsToRun.push(node) - } - } + if (sessionsFoundAtWorkspace !== undefined) { // it might be undefined when you are spawning a session that was running while backed up + sessionsFoundAtWorkspace.push(node) } } } diff --git a/CanvasApp/DesignSpace/Workspace/Workspace.js b/CanvasApp/DesignSpace/Workspace/Workspace.js index 747baf496..705a02555 100644 --- a/CanvasApp/DesignSpace/Workspace/Workspace.js +++ b/CanvasApp/DesignSpace/Workspace/Workspace.js @@ -122,7 +122,7 @@ function newWorkspace() { function finishInitialization() { setupEventsServerClients() - runTasksAndSessions(false) + runTasksAndSessions() thisObject.enabled = true canvas.cockpitSpace.initializePosition() canvas.splashScreen.initialize() @@ -134,21 +134,13 @@ function newWorkspace() { } } - function runTasksAndSessions(replacingCurrentWorkspace) { - if (replacingCurrentWorkspace === true) { - // We need to wait all tasks that were potentially running to stop - setTimeout(functionLibraryUiObjectsFromNodes.runTasks, 70000) - // We give a few seconds for the tasks to start - setTimeout(functionLibraryUiObjectsFromNodes.runSessions, 80000) - // We give a few seconds - setTimeout(functionLibraryUiObjectsFromNodes.playTutorials, 3000) - } else { - functionLibraryUiObjectsFromNodes.runTasks() - // We give a few seconds for the tasks to start - setTimeout(functionLibraryUiObjectsFromNodes.runSessions, 10000) - // We give a few seconds - setTimeout(functionLibraryUiObjectsFromNodes.playTutorials, 1000) - } + function runTasksAndSessions() { + + functionLibraryUiObjectsFromNodes.syncronizeTasksFoundAtWorkspaceWithBackEnd(functionLibraryTaskFunctions) + functionLibraryUiObjectsFromNodes.syncronizeSessionsFoundAtWorkspaceWithBackEnd(functionLibrarySessionFunctions) + + setTimeout(functionLibraryUiObjectsFromNodes.playTutorials, 5000) + } function setupEventsServerClients() { @@ -294,9 +286,11 @@ function newWorkspace() { workingAtTask++ break case 5: - functionLibraryUiObjectsFromNodes.recreateWorkspace(thisObject.workspaceNode) - setupEventsServerClients() - runTasksAndSessions(true) + functionLibraryUiObjectsFromNodes.recreateWorkspace(thisObject.workspaceNode, finishInitialization) + function finishInitialization() { + setupEventsServerClients() + runTasksAndSessions() + } workingAtTask++ break case 6: diff --git a/CanvasApp/FloatingSpace/UiObject.js b/CanvasApp/FloatingSpace/UiObject.js index dfef9b09f..91a22d8c0 100644 --- a/CanvasApp/FloatingSpace/UiObject.js +++ b/CanvasApp/FloatingSpace/UiObject.js @@ -980,7 +980,7 @@ function newUiObject() { /* We will wait to hear the Running event in order to confirm the execution was really started */ let key = thisObject.payload.node.name + '-' + thisObject.payload.node.type + '-' + thisObject.payload.node.id eventsServerClient.listenToEvent(key, 'Running', undefined, 'UiObject', onResponse, onRunning) - + onRunningCallBackFunction = callBackFunction function onResponse(message) { @@ -988,6 +988,7 @@ function newUiObject() { } function onRunning() { + if (thisObject.payload === undefined) { return } let key = thisObject.payload.node.name + '-' + thisObject.payload.node.type + '-' + thisObject.payload.node.id diff --git a/TaskServer/LowFrequencyTrading/TradingProcess.js b/TaskServer/LowFrequencyTrading/TradingProcess.js index b6298e6f0..5a37df7c6 100644 --- a/TaskServer/LowFrequencyTrading/TradingProcess.js +++ b/TaskServer/LowFrequencyTrading/TradingProcess.js @@ -146,6 +146,8 @@ } else { /* We are processing Daily Files */ do { + global.EMIT_SESSION_STATUS (bot.SESSION_STATUS, bot.sessionKey) + if (checkStopTaskGracefully() === false) { break } if (checkStopProcessing() === false) { break } @@ -401,7 +403,7 @@ async function processDailyFiles() { /* Telling the world we are alive and doing well and which date we are processing right now. */ let processingDateString = tradingProcessDate.getUTCFullYear() + '-' + utilities.pad(tradingProcessDate.getUTCMonth() + 1, 2) + '-' + utilities.pad(tradingProcessDate.getUTCDate(), 2); - bot.processHeartBeat(processingDateString) + bot.processHeartBeat(processingDateString, undefined, "Running...") /* We will iterate through all posible timeFrames. diff --git a/TaskServer/Session.js b/TaskServer/Session.js index 64b4e0c30..926fbcc09 100644 --- a/TaskServer/Session.js +++ b/TaskServer/Session.js @@ -38,8 +38,9 @@ bot.sessionKey = bot.processNode.session.name + '-' + bot.processNode.session.type + '-' + bot.processNode.session.id global.SESSION_MAP.set(bot.sessionKey, bot.sessionKey) - global.EVENT_SERVER_CLIENT.listenToEvent(bot.sessionKey, 'Run Session', undefined, bot.sessionKey, undefined, onSessionRan) - global.EVENT_SERVER_CLIENT.listenToEvent(bot.sessionKey, 'Stop Session', undefined, bot.sessionKey, undefined, onSessionStopped) + global.EVENT_SERVER_CLIENT.listenToEvent(bot.sessionKey, 'Session Status', undefined, bot.sessionKey, undefined, onSessionStatus) + global.EVENT_SERVER_CLIENT.listenToEvent(bot.sessionKey, 'Run Session', undefined, bot.sessionKey, undefined, onSessionRun) + global.EVENT_SERVER_CLIENT.listenToEvent(bot.sessionKey, 'Stop Session', undefined, bot.sessionKey, undefined, onSessionStop) /* Connect this here so that it is accesible from other places */ bot.sessionError = sessionError @@ -52,11 +53,25 @@ callBackFunction(global.DEFAULT_OK_RESPONSE) return - function onSessionRan(message) { + function onSessionStatus() { + if (bot.SESSION_STATUS === 'Running') { + let event = { + status: 'Session Runnning' + } + global.EVENT_SERVER_CLIENT.raiseEvent(bot.sessionKey, 'Status Response', event) + } else { + let event = { + status: 'Not Session Runnning' + } + global.EVENT_SERVER_CLIENT.raiseEvent(bot.sessionKey, 'Status Response', event) + } + } + + function onSessionRun(message) { try { /* This happens when the UI is reloaded, the session was running and tries to run it again. */ if (bot.SESSION_STATUS === 'Idle' || bot.SESSION_STATUS === 'Running') { - parentLogger.write(MODULE_NAME, "[WARN] onSessionRan -> Event receive to run the Session while it was already running. ") + parentLogger.write(MODULE_NAME, "[WARN] onSessionRun -> Event receive to run the Session while it was already running. ") return } @@ -100,16 +115,16 @@ bot.STOP_SESSION = false } else { bot.STOP_SESSION = true - if (FULL_LOG === true) { parentLogger.write(MODULE_NAME, '[IMPORTANT] onSessionRan -> Stopping the Session now. ') } + if (FULL_LOG === true) { parentLogger.write(MODULE_NAME, '[IMPORTANT] onSessionRun -> Stopping the Session now. ') } } socialBotsModule.sendMessage(bot.SESSION.type + " '" + bot.SESSION.name + "' is starting.") } catch (err) { - parentLogger.write(MODULE_NAME, "[ERROR] initialize -> onSessionRan -> err = " + err.stack); + parentLogger.write(MODULE_NAME, "[ERROR] initialize -> onSessionRun -> err = " + err.stack); } } - function onSessionStopped() { + function onSessionStop() { stopSession('Session Stopped From the User Interface.') } diff --git a/TaskServer/TradingBot.js b/TaskServer/TradingBot.js index e8c03c194..9d4d48ee7 100644 --- a/TaskServer/TradingBot.js +++ b/TaskServer/TradingBot.js @@ -74,7 +74,7 @@ let nextWaitTime; /* We tell the UI that we are running. */ - processHeartBeat() + processHeartBeat(undefined, undefined, "Running...") /* We define here all the modules that the rest of the infraestructure, including the bots themselves can consume. */ const UTILITIES = require(global.ROOT_DIR + 'CloudUtilities'); @@ -94,14 +94,14 @@ /* Checking if we need to need to emit any event */ if (bot.SESSION_STATUS === 'Idle' && bot.STOP_SESSION === false) { - global.EVENT_SERVER_CLIENT.raiseEvent(bot.sessionKey, 'Running') bot.SESSION_STATUS = 'Running' } if (bot.SESSION_STATUS === 'Running' && bot.STOP_SESSION === true) { - global.EVENT_SERVER_CLIENT.raiseEvent(bot.sessionKey, 'Stopped') bot.SESSION_STATUS = 'Stopped' - } + } + + global.EMIT_SESSION_STATUS (bot.SESSION_STATUS, bot.sessionKey) /* Checking if we should process this loop or not.*/ if (bot.STOP_SESSION === true) { @@ -523,7 +523,7 @@ if (FULL_LOG === true) { logger.write(MODULE_NAME, "[INFO] run -> loop -> loopControl -> nextWaitTime = " + nextWaitTime); } /* We show we reached the end of the loop. */ - processHeartBeat() + processHeartBeat(undefined, undefined, "Running...") /* Here we check if we must stop the loop gracefully. */ shallWeStop(onStop, onContinue); diff --git a/TaskServer/server.js b/TaskServer/server.js index 754888678..3ff27cf3f 100644 --- a/TaskServer/server.js +++ b/TaskServer/server.js @@ -285,6 +285,18 @@ global.FILTER_OUT_NODES_WITHOUT_REFERENCE_PARENT_FROM_NODE_ARRAY = function (nod return filteredNodeArray } +global.EMIT_SESSION_STATUS = function (status, key) { + switch(status) { + case 'Running': { + global.EVENT_SERVER_CLIENT.raiseEvent(key, 'Running') + break + } + case 'Stopped': { + global.EVENT_SERVER_CLIENT.raiseEvent(key, 'Stopped') + break + } + } +} process.on('uncaughtException', function (err) { console.log('[ERROR] Task Server -> server -> uncaughtException -> err.message = ' + err.message)