fez5587/Superalgos
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Session and Task Syncronization with the backend

Browse Source
This commit is contained in:
Luis Fernando Molina committed 2020-09-25 20:24:32 +02:00
1 parent 1c2a335f6f
commit ec4450f0fe
10 files changed
+115 -75
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

BackendServers/taskManagerServer.js
+4 -5
View File
@@ -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)
}
}
}
CanvasApp/DesignSpace/Workspace/FunctionLibraries/SessionFunctions.js
+30 -5
View File
@@ -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
CanvasApp/DesignSpace/Workspace/FunctionLibraries/TaskFunctions.js
+3 -3
View File
@@ -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')
}
}
CanvasApp/DesignSpace/Workspace/FunctionLibraries/UiObjectsFromNodes.js
+21 -29
View File
@@ -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)
}
}
}
CanvasApp/DesignSpace/Workspace/Workspace.js
+13 -19
View File
@@ -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:
CanvasApp/FloatingSpace/UiObject.js
+2 -1
View File
@@ -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
TaskServer/LowFrequencyTrading/TradingProcess.js
+3 -1
View File
@@ -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.
TaskServer/Session.js
+22 -7
View File
@@ -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.')
}
TaskServer/TradingBot.js
+5 -5
View File
@@ -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);
TaskServer/server.js
+12
View File
@@ -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)