From e07273d026c9b3ee35f8a7a9c0ad21a744807f91 Mon Sep 17 00:00:00 2001 From: Luis Fernando Molina Date: Sat, 4 Apr 2020 15:40:15 +0200 Subject: [PATCH 1/5] Update server.js --- server.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/server.js b/server.js index d426bd320..746f895ad 100644 --- a/server.js +++ b/server.js @@ -72,7 +72,7 @@ function runTask(message) { id: message.event.taskId, name: message.event.taskName } - + tasks.set(message.event.taskId, task) task.childProcess.on('error', (err) => { @@ -86,7 +86,7 @@ function runTask(message) { tasks.delete(task.id) }) - setTimeout(sendStartingEvent, 3000) + global.SYSTEM_EVENT_HANDLER.listenToEvent('Task Manager - ' + message.event.taskId, 'Nodejs Process Ready for Task', undefined, undefined, undefined, sendStartingEvent) function sendStartingEvent() { //console.log('[INFO] Task Manager -> server -> runTask -> Emitting Event -> key = ' + 'Task Server - ' + task.id) From eb79edefdef28e256551952306de902bf07c5e8d Mon Sep 17 00:00:00 2001 From: Luis Fernando Molina Date: Mon, 6 Apr 2020 08:06:47 +0200 Subject: [PATCH 2/5] Removed IPC, changed it for Web Sockets --- TaskManager.njsproj => BackendServers.njsproj | 11 +- TaskManager.sln => BackendServers.sln | 7 +- SystemEventHandler.js | 178 ----------- eventsServer.js | 223 +++++++++++++ package.json | 4 +- server.js | 149 +++------ taskManagerServer.js | 295 ++++++++++++++++++ webSocketsServer.js | 112 +++++++ 8 files changed, 686 insertions(+), 293 deletions(-) rename TaskManager.njsproj => BackendServers.njsproj (87%) rename TaskManager.sln => BackendServers.sln (73%) delete mode 100644 SystemEventHandler.js create mode 100644 eventsServer.js create mode 100644 taskManagerServer.js create mode 100644 webSocketsServer.js diff --git a/TaskManager.njsproj b/BackendServers.njsproj similarity index 87% rename from TaskManager.njsproj rename to BackendServers.njsproj index 301f1ed74..d19babbd5 100644 --- a/TaskManager.njsproj +++ b/BackendServers.njsproj @@ -2,7 +2,7 @@ 14.0 $(MSBuildExtensionsPath32)\Microsoft\VisualStudio\v$(VisualStudioVersion) - TaskManager + BackendServers BackendServer @@ -29,7 +29,16 @@ true + + Code + + + Code + + + Code + diff --git a/TaskManager.sln b/BackendServers.sln similarity index 73% rename from TaskManager.sln rename to BackendServers.sln index 33ecf3244..5a65b7364 100644 --- a/TaskManager.sln +++ b/BackendServers.sln @@ -3,7 +3,12 @@ Microsoft Visual Studio Solution File, Format Version 12.00 # Visual Studio 15 VisualStudioVersion = 15.0.27130.2027 MinimumVisualStudioVersion = 10.0.40219.1 -Project("{9092AA53-FB77-4645-B42D-1CCCA6BD08BD}") = "TaskManager", "TaskManager.njsproj", "{6A53AD2E-C389-4D2C-BDED-23DBAB231315}" +Project("{9092AA53-FB77-4645-B42D-1CCCA6BD08BD}") = "BackendServers", "BackendServers.njsproj", "{6A53AD2E-C389-4D2C-BDED-23DBAB231315}" +EndProject +Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{BEEFD357-76B2-431F-A1A9-045C52E7F2C6}" + ProjectSection(SolutionItems) = preProject + .env = .env + EndProjectSection EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution diff --git a/SystemEventHandler.js b/SystemEventHandler.js deleted file mode 100644 index 7620ca72e..000000000 --- a/SystemEventHandler.js +++ /dev/null @@ -1,178 +0,0 @@ -exports.newSystemEventHandler = function newSystemEventHandler(ipc) { - let thisObject = { - initialize: initialize, - finalize: finalize, - createEventHandler: createEventHandler, - deleteEventHandler: deleteEventHandler, - listenToEvent: listenToEvent, - stopListening: stopListening, - raiseEvent: raiseEvent - } - - let eventListeners = new Map() - let responseWaiters = new Map() - - ipc.config.id = 'Task Server'; - ipc.config.retry = 1500; - ipc.config.silent = true; - - return thisObject - - function initialize(canllerName, callBackFunction) { - - ipc.connectTo( - 'world', - function () { - ipc.of.world.on( - 'connect', - function () { - ipc.log('## Connected to Superalgos Event Server ##'.rainbow, ipc.config.delay); - - let eventCommand = { - action: 'greetings', - from: canllerName - } - - sendCommand(eventCommand) - - callBackFunction() - } - ); - ipc.of.world.on( - 'disconnect', - function () { - ipc.log('Disconnected from Superalgos Event Server '.notice); - } - ); - ipc.of.world.on( - 'message', // here is where all incomming messages are processed. - function (data) { - ipc.log('Got a message from Superalgos Event Server: '.debug, data); - - let message = JSON.parse(data) - - if (message.action === 'Event Raised') { - let key - if (message.callerId) { - key = message.eventHandlerName + '-' + message.eventType + '-' + message.callerId - } else { - key = message.eventHandlerName + '-' + message.eventType - } - let handler = eventListeners.get(key) - if (handler) { - handler.callBack(message) - } else { - console.log(key + ' not found so could not deliver event raised.') - console.log(' Message = ' + data) - } - return - } - - if (message.action === 'Event Server Response') { - let handler = responseWaiters.get(message.callerId) - if (handler) { - handler(message) - } - return - } - } - ); - } - ); - } - - function finalize() { - /* Before disconnecting we will forze all eventListeners to stop listening. */ - const eventListenersArray = [...eventListeners.values()] - for (let i = 0; i < eventListenersArray.length; i++) { - let handler = eventListenersArray[i] - let eventCommand = { - action: 'stopListening', - eventHandlerName: handler.eventHandlerName, - eventType: handler.eventType, - callerId: handler.callerId - } - sendCommand(eventCommand) - } - ipc.disconnect('world'); - } - - function sendCommand(command, responseCallBack, eventsCallBack) { - - if (command.action === 'listenToEvent') { - let key - if (command.callerId) { - key = command.eventHandlerName + '-' + command.eventType + '-' + command.callerId - } else { - key = command.eventHandlerName + '-' + command.eventType - } - let handler = { - eventHandlerName: command.eventHandlerName, - eventType: command.eventType, - callerId: command.callerId, - callBack: eventsCallBack - } - eventListeners.set(key, handler) - } - if (command.callerId && responseCallBack) { - responseWaiters.set(command.callerId, responseCallBack) - } - - ipc.of.world.emit( - 'message', - JSON.stringify(command) - ) - } - - function createEventHandler(eventHandlerName, callerId, responseCallBack) { - let eventCommand = { - action: 'createEventHandler', - eventHandlerName: eventHandlerName, - callerId: callerId - } - sendCommand(eventCommand, responseCallBack) - } - - function deleteEventHandler(eventHandlerName, callerId, responseCallBack) { - let eventCommand = { - action: 'deleteEventHandler', - eventHandlerName: eventHandlerName, - callerId: callerId - } - sendCommand(eventCommand, responseCallBack) - } - - function listenToEvent(eventHandlerName, eventType, extraData, callerId, responseCallBack, eventsCallBack) { - let eventCommand = { - action: 'listenToEvent', - eventHandlerName: eventHandlerName, - eventType: eventType, - extraData: extraData, - callerId: callerId - } - sendCommand(eventCommand, responseCallBack, eventsCallBack) - } - - function stopListening(eventHandlerName, eventType, eventSubscriptionId, callerId, responseCallBack) { - /* User either needs to specify a valid eventSubscriptionId OR the 3 params: eventHandlerName, eventType, callerId which were used when start listening to events. */ - let eventCommand = { - action: 'stopListening', - eventHandlerName: eventHandlerName, - eventType: eventType, - eventSubscriptionId: eventSubscriptionId, - callerId: callerId - } - sendCommand(eventCommand, responseCallBack) - } - - function raiseEvent(eventHandlerName, eventType, event, callerId, responseCallBack) { - let eventCommand = { - action: 'raiseEvent', - eventHandlerName: eventHandlerName, - eventType: eventType, - event: event, - callerId: callerId - } - sendCommand(eventCommand, responseCallBack) - } -} \ No newline at end of file diff --git a/eventsServer.js b/eventsServer.js new file mode 100644 index 000000000..2f8fc9bb4 --- /dev/null +++ b/eventsServer.js @@ -0,0 +1,223 @@ +exports.newEventsServer = function newEventsServer() { + + const MODULE = "Events Server" + const LOG_INFO = false + + let thisObject = { + initialize: initialize, + finalize: finalize, + run: run, + onMessage: onMessage + } + + let eventHandlers = new Map() + let counter = 0 + + return thisObject + + function finalize() { + + } + + function initialize() { + + } + + function onMessage(data, callbackFunction) { + try { + let command = JSON.parse(data) + + switch (command.action) { + case "greetings": { + sendResponse(global.DEFAULT_OK_RESPONSE) + return + } + case "createEventHandler": { + let eventHandler = eventHandlers.get(command.eventHandlerName) + if (eventHandler === undefined) { + /* + We will only create event handlers which were not created before. Remember that someone trying to listen events + at an event handler that does not exist, will create that eventhandler, so it is possible that the emitter + when trying to create the event handler in fact was late and the event handler is already there if the listener + arrived here first. An event handler created with this method is not destroyed if there are no listeners or they + all leave. + */ + eventHandler = newEventHandler() + eventHandler.deleteWhenAllListenersAreGone = false + eventHandlers.set(command.eventHandlerName, eventHandler) + } + sendResponse(global.DEFAULT_OK_RESPONSE) + return + } + case "deleteEventHandler": { + let eventHandler = eventHandlers.get(command.eventHandlerName) + if (eventHandler !== undefined) { + /* + We will only delete event handlers which have noone listening, allowing a listener to continue listening even + when the event emitter is gone, asuming that it could come back at some point in time and resume emitting. + */ + if (eventHandler.listeners.length === 0) { + eventHandlers.delete(command.eventHandlerName) + } else { + eventHandler.deleteWhenAllListenersAreGone = true + } + } + + sendResponse(global.DEFAULT_OK_RESPONSE) + return + } + case "listenToEvent": { + let eventHandler = eventHandlers.get(command.eventHandlerName) + if (eventHandler === undefined) { + /* + We will create event handlers which were not created before if a listener tries to listen to a handler that does noet + exist before the emitter comes and create it. This is to aviod syncronization problems and also problems with emitters life + cycles with higher frecuency that the ones of listeners. Created in this way, event handlers are marked for deletion if + all listeners stop listening at one point in time. + */ + eventHandler = newEventHandler() + eventHandler.deleteWhenAllListenersAreGone = true + eventHandlers.set(command.eventHandlerName, eventHandler) + } + eventSubscriptionId = eventHandler.listenToEvent(command.eventHandlerName, command.eventType, command.callerId, handlerFunction, command.extraData) + + function handlerFunction(event, extraData) { + let outputData = { + action: 'Event Raised', + eventHandlerName: command.eventHandlerName, + eventType: command.eventType, + event: event, + callerId: command.callerId, + extraData: extraData + } + callbackFunction(JSON.stringify(outputData)) + } + + let response = { + result: global.DEFAULT_OK_RESPONSE.result, + message: global.DEFAULT_OK_RESPONSE.message, + eventSubscriptionId: eventSubscriptionId + } + + sendResponse(response) + return + } + case "stopListening": { + let eventHandler = eventHandlers.get(command.eventHandlerName) + + if (eventHandler === undefined) { + + console.log("[WARN] BackendServers -> Events Server -> onMessage -> Cannot to stop listening to events at an Event Handler that does not exist. Command = " + JSON.stringify(command)) + + sendResponse(global.DEFAULT_FAIL_RESPONSE) + return + } + + eventHandler.stopListening(command.eventHandlerName, command.eventType, command.callerId, command.eventSubscriptionId) + + /* + We check here if there are no more listeners and the event handler original emmiter is also gone, then we need to delete + this event handlers since chances are that is not needed anymore. + */ + if (eventHandler.listeners.length === 0 && eventHandler.deleteWhenAllListenersAreGone === true) { + eventHandlers.delete(command.eventHandlerName) + } + + sendResponse(global.DEFAULT_OK_RESPONSE) + return + } + case "raiseEvent": { + let eventHandler = eventHandlers.get(command.eventHandlerName) + + if (eventHandler === undefined) { + + eventHandler = newEventHandler() + eventHandlers.set(command.eventHandlerName, eventHandler) + + /* No matter if it is the listener or the emiter the one that acts first, if an event handler does not exist we just create it acan keep it there. */ + + } + //console.log("[INFO] BackendServers -> Events Server -> Raising Event " + command.eventHandlerName + " " + command.eventType) + eventHandler.raiseEvent(command.eventType, command.event) + sendResponse(global.DEFAULT_OK_RESPONSE) + return + } + } + + console.log("[WARN] BackendServers -> Events Server -> onMessage -> Unknown Command Received:" + data) + + function sendResponse(message) { + if (command.callerId !== undefined) { + message.callerId = command.callerId + } + message.action = 'Event Server Response' + callbackFunction(JSON.stringify(message)) + } + + } catch (err) { + console.log("[ERROR] BackendServers -> Events Server -> onMessage -> Bad Command Received:" + data) + console.log("[ERROR] BackendServers -> Events Server -> onMessage -> An Error Happened:" + err.stack) + } + } + + function run() { + console.log('Events Server Started.') + } + + function newEventHandler() { + let thisObject = { + name: undefined, // This is for debugging purposes only. + deleteWhenAllListenersAreGone: false, + listeners: [], // Here we store all the functions we will call when an event is raiseed. + listenToEvent: listenToEvent, + stopListening: stopListening, + raiseEvent: raiseEvent, + finalize: finalize + } + + return thisObject + + function finalize() { + thisObject.listeners = [] + } + + function listenToEvent(eventHandlerName, eventType, callerId, handler, extraData) { + let eventSubscriptionId = Math.trunc(Math.random() * 1000000) + thisObject.listeners.push([eventHandlerName, eventType, callerId, handler, extraData, eventSubscriptionId]) + return eventSubscriptionId + } + + function stopListening(eventHandlerName, eventType, callerId, eventSubscriptionId) { + + if (eventSubscriptionId !== undefined) { + for (let i = 0; i < thisObject.listeners.length; i++) { + if (eventSubscriptionId === thisObject.listeners[i][5]) { + thisObject.listeners.splice(i, 1) + return + } + } + } + if (eventHandlerName !== undefined && eventType !== undefined && callerId !== undefined) { + for (let i = 0; i < thisObject.listeners.length; i++) { + if (eventHandlerName === thisObject.listeners[i][0] && eventType === thisObject.listeners[i][1] && callerId === thisObject.listeners[i][2]) { + thisObject.listeners.splice(i, 1) + return + } + } + } + } + + function raiseEvent(eventType, event) { + for (let i = 0; i < thisObject.listeners.length; i++) { + /* We will execute all the functions listening to this event type. */ + + if (thisObject.listeners[i][1] === eventType) { + let handler = thisObject.listeners[i][3] + let extraData = thisObject.listeners[i][4] + + handler(event, extraData) + } + } + } + } +} \ No newline at end of file diff --git a/package.json b/package.json index 567884cf5..37c6ecb63 100644 --- a/package.json +++ b/package.json @@ -8,7 +8,7 @@ }, "dependencies": { "child_process": "^1.0.2", - "dotenv": "^8.1.0", - "node-ipc": "^9.1.1" + "dotenv": "^8.2.0", + "ws": "^7.1.2" } } diff --git a/server.js b/server.js index 746f895ad..a9636899b 100644 --- a/server.js +++ b/server.js @@ -1,122 +1,49 @@ -/* -This server creates and destroys Task Servers according to what is going on at the UI. Each of these created servers will do their own -work. The responsibility of this server is to manage those children processes in order to be in sync with what the end user want and is defining -at the user interface. +process.on('uncaughtException', function (err) { + if (err.message.indexOf("EADDRINUSE") > 0) { + console.log("Superalgos Backend Server cannot be started. Reason: the port " + port + " is already in use by another application.") + return + } + console.log('[ERROR] Backend Server -> server -> uncaughtException -> err.message = ' + err.message) + console.log('[ERROR] Backend Server -> server -> uncaughtException -> err.stack = ' + err.stack) + process.exit(1) +}) -*/ +process.on('unhandledRejection', (reason, p) => { + console.log('[ERROR] Backend Server -> server -> unhandledRejection -> reason = ' + JSON.stringify(reason)) + console.log('[ERROR] Backend Server -> server -> unhandledRejection -> p = ' + JSON.stringify(p)) + process.exit(1) +}) -const { fork } = require('child_process') -require('dotenv').config() -tasks = new Map() +/* Callbacks default responses. */ -const EVENT_HANDLER_MODULE = require('./SystemEventHandler.js'); -const IPC = require('node-ipc'); -global.SYSTEM_EVENT_HANDLER = EVENT_HANDLER_MODULE.newSystemEventHandler(IPC) -global.SYSTEM_EVENT_HANDLER.initialize('Task Manager', bootLoader) +global.DEFAULT_OK_RESPONSE = { + result: "Ok", + message: "Operation Succeeded" +}; -console.log('Task Manager Server Started.') +global.DEFAULT_FAIL_RESPONSE = { + result: "Fail", + message: "Operation Failed" +}; -/* +let EVENTS_SERVER = require('./eventsServer.js'); +let WEB_SOCKETS_SERVER = require('./webSocketsServer.js'); +let TASK_MANAGER_SERVER = require('./taskManagerServer.js'); -Here we will create an Event Handler for this server. After that we will listen to events happening here. Elements of the UI will raise events here, -as a way to communicate what is going on with the user interacting with UI components. That is the way this server will know for example when a new -task needs to be created or destroyed. +try { + EVENTS_SERVER = EVENTS_SERVER.newEventsServer() + EVENTS_SERVER.initialize() + EVENTS_SERVER.run() -*/ + WEB_SOCKETS_SERVER = WEB_SOCKETS_SERVER.newWebSocketsServer(EVENTS_SERVER) + WEB_SOCKETS_SERVER.initialize() + WEB_SOCKETS_SERVER.run() -function bootLoader() { - global.SYSTEM_EVENT_HANDLER.createEventHandler('Task Manager') - global.SYSTEM_EVENT_HANDLER.listenToEvent('Task Manager', 'Run Task', undefined, undefined, undefined, runTask) - global.SYSTEM_EVENT_HANDLER.listenToEvent('Task Manager', 'Stop Task', undefined, undefined, undefined, stopTask) + TASK_MANAGER_SERVER = TASK_MANAGER_SERVER.newTaskManagerServer(WEB_SOCKETS_SERVER, EVENTS_SERVER) + TASK_MANAGER_SERVER.initialize() + TASK_MANAGER_SERVER.run() +} catch (err) { + console.log('[ERROR] BackendServers -> Task Manager -> server -> Error = ' + err.stack) } -function runTask(message) { - //console.log('[INFO] Task Manager -> server -> runTask -> Entering function.') - - if (message.event === undefined) { - console.log('[WARN] Task Manager -> server -> runTask -> Message Received Without Event -> message = ' + JSON.stringify(message)) - return - } - - if (message.event.taskId === undefined) { - console.log('[WARN] Task Manager -> server -> runTask -> Message Received Without taskId -> message = ' + JSON.stringify(message)) - return - } - - if (message.event.definition === undefined) { - console.log('[WARN] Task Manager -> server -> runTask -> Message Received Without Definition -> message = ' + JSON.stringify(message)) - return - } - - if (tasks.get(message.event.taskId) !== undefined) { - //console.log('[WARN] Task Manager -> server -> runTask -> Task Already Running -> taskId = ' + message.event.taskId) - return - } - //console.log('[INFO] Task Manager -> server -> runTask -> Task Name = ' + message.event.taskName) - //console.log('[INFO] Task Manager -> server -> runTask -> Task Id = ' + message.event.taskId) - - let path = process.env.TASK_SERVER_PATH + '/server.js' - - /* Workarround to avoid having the same debug port at the forked process which makes it crash. */ - for (let i = 0; i < process.execArgv.length; i++) { - let argument = process.execArgv[i] - if (argument.indexOf('--inspect') > -1) { - process.execArgv[i] = '' - } - } - - /* Forking this process. */ - let task = { - childProcess: fork(path, [message.event.taskId], { stdio: 'inherit' }), - id: message.event.taskId, - name: message.event.taskName - } - - tasks.set(message.event.taskId, task) - - task.childProcess.on('error', (err) => { - console.log('[ERROR] Task Manager -> server -> runTask -> Problem with Task Name = ' + task.name) - console.log('[ERROR] Task Manager -> server -> runTask -> Problem with Task Id = ' + task.id) - console.log(`[ERROR] Task Manager -> server -> runTask -> Task Manager exited with error ${err}`) - }) - task.childProcess.on('close', (code, signal) => { - //console.log('[INFO] Task Manager -> server -> runTask -> Task Terminated. -> Task Name = ' + task.name) - //console.log('[INFO] Task Manager -> server -> runTask -> Task Terminated. -> Task Id = ' + task.id) - tasks.delete(task.id) - }) - - global.SYSTEM_EVENT_HANDLER.listenToEvent('Task Manager - ' + message.event.taskId, 'Nodejs Process Ready for Task', undefined, undefined, undefined, sendStartingEvent) - - function sendStartingEvent() { - //console.log('[INFO] Task Manager -> server -> runTask -> Emitting Event -> key = ' + 'Task Server - ' + task.id) - global.SYSTEM_EVENT_HANDLER.raiseEvent('Task Server - ' + task.id, 'Run Task', message.event) - } -} - -function stopTask(message) { - - //console.log('[INFO] Task Manager -> server -> stopTask -> Entering function.') - - if (message.event === undefined) { - console.log('[WARN] Task Manager -> server -> stopTask -> Message Received Without Event -> message = ' + JSON.stringify(message)) - return - } - - if (message.event.taskId === undefined) { - console.log('[WARN] Task Manager -> server -> stopTask -> Message Received Without taskId -> message = ' + JSON.stringify(message)) - return - } - - //console.log('[INFO] Task Manager -> server -> stopTask -> Task Name = ' + message.event.taskName) - //console.log('[INFO] Task Manager -> server -> stopTask -> Task Id = ' + message.event.taskId) - - let task = tasks.get(message.event.taskId) - - if (task) { - task.childProcess.send('Stop this Task'); - //console.log('[INFO] Task Manager -> server -> stopTask -> Child Process instructed to finish.') - } else { - console.log('[WARN] Task Manager -> server -> stopTask -> Cannot delete Task that does not exist.') - } -} \ No newline at end of file diff --git a/taskManagerServer.js b/taskManagerServer.js new file mode 100644 index 000000000..72b1d3b24 --- /dev/null +++ b/taskManagerServer.js @@ -0,0 +1,295 @@ +exports.newTaskManagerServer = function newTaskManagerServer(WEB_SOCKETS_SERVER, EVENTS_SERVER) { + + const MODULE = "Task Manager Server" + + let thisObject = { + initialize: initialize, + finalize: finalize, + run: run + } + + /* + + This server creates and destroys Task Servers according to what is going on at the UI. Each of these created servers will do their own + work. The responsibility of this server is to manage those children processes in order to be in sync with what the end user want and is defining + at the user interface. + + */ + + const { fork } = require('child_process') + require('dotenv').config() + tasks = new Map() + + return thisObject + + function finalize() { + + } + + function initialize() { + + } + + function run() { + + /* + + Here we will create an Event Handler for this server. After that we will listen to events happening here. Elements of the UI will raise events here, + as a way to communicate what is going on with the user interacting with UI components. That is the way this server will know for example when a new + task needs to be created or destroyed. + + */ + + let eventsServerClient = newEventsServerClient() + eventsServerClient.createEventHandler('Task Manager') + eventsServerClient.listenToEvent('Task Manager', 'Run Task', undefined, undefined, undefined, runTask) + eventsServerClient.listenToEvent('Task Manager', 'Stop Task', undefined, undefined, undefined, stopTask) + + console.log('Task Manager Server Started.') + + function runTask(message) { + //console.log('[INFO] BackendServers -> Task Manager Server -> runTask -> Entering function.') + + if (message.event === undefined) { + console.log('[WARN] BackendServers -> Task Manager Server -> runTask -> Message Received Without Event -> message = ' + JSON.stringify(message)) + return + } + + if (message.event.taskId === undefined) { + console.log('[WARN] BackendServers -> Task Manager Server -> runTask -> Message Received Without taskId -> message = ' + JSON.stringify(message)) + return + } + + if (message.event.definition === undefined) { + console.log('[WARN] BackendServers -> Task Manager Server -> runTask -> Message Received Without Definition -> message = ' + JSON.stringify(message)) + return + } + + if (tasks.get(message.event.taskId) !== undefined) { + //console.log('[WARN] BackendServers -> Task Manager Server -> runTask -> Task Already Running -> taskId = ' + message.event.taskId) + return + } + //console.log('[INFO] BackendServers -> Task Manager Server -> runTask -> Task Name = ' + message.event.taskName) + //console.log('[INFO] BackendServers -> Task Manager Server -> runTask -> Task Id = ' + message.event.taskId) + + let path = process.env.TASK_SERVER_PATH + '/server.js' + + /* Workarround to avoid having the same debug port at the forked process which makes it crash. */ + for (let i = 0; i < process.execArgv.length; i++) { + let argument = process.execArgv[i] + if (argument.indexOf('--inspect') > -1) { + process.execArgv[i] = '' + } + } + + /* Forking this process. */ + let task = { + childProcess: fork(path, [message.event.taskId], { stdio: 'inherit' }), + id: message.event.taskId, + name: message.event.taskName + } + + tasks.set(message.event.taskId, task) + + task.childProcess.on('error', (err) => { + console.log('[ERROR] BackendServers -> Task Manager Server -> runTask -> Problem with Task Name = ' + task.name) + console.log('[ERROR] BackendServers -> Task Manager Server -> runTask -> Problem with Task Id = ' + task.id) + console.log(`[ERROR] BackendServers -> Task Manager Server -> runTask -> Task Manager exited with error ${err}`) + }) + task.childProcess.on('close', (code, signal) => { + //console.log('[INFO] BackendServers -> Task Manager Server -> runTask -> Task Terminated. -> Task Name = ' + task.name) + //console.log('[INFO] BackendServers -> Task Manager Server -> runTask -> Task Terminated. -> Task Id = ' + task.id) + tasks.delete(task.id) + }) + + eventsServerClient.listenToEvent('Task Manager - ' + message.event.taskId, 'Nodejs Process Ready for Task', undefined, undefined, undefined, sendStartingEvent) + + function sendStartingEvent() { + //console.log('[INFO] BackendServers -> Task Manager Server -> runTask -> Emitting Event -> key = ' + 'Task Server - ' + task.id) + eventsServerClient.raiseEvent('Task Server - ' + task.id, 'Run Task', message.event) + } + } + + function stopTask(message) { + + //console.log('[INFO] BackendServers -> Task Manager Server -> stopTask -> Entering function.') + + if (message.event === undefined) { + console.log('[WARN] BackendServers -> Task Manager Server -> stopTask -> Message Received Without Event -> message = ' + JSON.stringify(message)) + return + } + + if (message.event.taskId === undefined) { + console.log('[WARN] BackendServers -> Task Manager Server -> stopTask -> Message Received Without taskId -> message = ' + JSON.stringify(message)) + return + } + + //console.log('[INFO] BackendServers -> Task Manager Server -> stopTask -> Task Name = ' + message.event.taskName) + //console.log('[INFO] BackendServers -> Task Manager Server -> stopTask -> Task Id = ' + message.event.taskId) + + let task = tasks.get(message.event.taskId) + + if (task) { + task.childProcess.send('Stop this Task'); + //console.log('[INFO] BackendServers -> Task Manager Server -> stopTask -> Child Process instructed to finish.') + } else { + console.log('[WARN] BackendServers -> Task Manager Server -> stopTask -> Cannot delete Task that does not exist.') + } + } + } + + function newEventsServerClient() { + let thisObject = { + initialize: initialize, + finalize: finalize, + createEventHandler: createEventHandler, + deleteEventHandler: deleteEventHandler, + listenToEvent: listenToEvent, + stopListening: stopListening, + raiseEvent: raiseEvent, + onMessage: onMessage + } + + let eventListeners = new Map() + let responseWaiters = new Map() + + return thisObject + + + function finalize() { + /* Before disconnecting we will forze all eventListeners to stop listening. */ + const eventListenersArray = [...eventListeners.values()] + for (let i = 0; i < eventListenersArray.length; i++) { + let handler = eventListenersArray[i] + let eventCommand = { + action: 'stopListening', + eventHandlerName: handler.eventHandlerName, + eventType: handler.eventType, + callerId: handler.callerId + } + sendCommand(eventCommand) + } + } + + function initialize() { + + } + + function onMessage(data) { + try { + let message = JSON.parse(data) + + if (message.action === 'Event Raised') { + let key + if (message.callerId) { + key = message.eventHandlerName + '-' + message.eventType + '-' + message.callerId + } else { + key = message.eventHandlerName + '-' + message.eventType + } + let handler = eventListeners.get(key) + if (handler) { + handler.callBack(message) + } else { + console.log(key + ' not found so could not deliver event raised.') + console.log(' Message = ' + data) + } + return + } + + if (message.action === 'Event Server Response') { + let handler = responseWaiters.get(message.callerId) + if (handler) { + handler(message) + } + return + } + } catch (err) { + console.log('[ERROR] BackendServers -> Task Manager Server -> onMessage -> Error Receiving Message from Events Server -> data = ' + JSON.stringify(data)) + console.log('[ERROR] BackendServers -> Task Manager Server -> onMessage -> Error Receiving Message from Events Server -> error = ' + err.stack) + } + } + + function sendCommand(command, responseCallBack, eventsCallBack) { + try { + if (command.action === 'listenToEvent') { + let key + if (command.callerId) { + key = command.eventHandlerName + '-' + command.eventType + '-' + command.callerId + } else { + key = command.eventHandlerName + '-' + command.eventType + } + let handler = { + eventHandlerName: command.eventHandlerName, + eventType: command.eventType, + callerId: command.callerId, + callBack: eventsCallBack + } + eventListeners.set(key, handler) + } + if (command.callerId && responseCallBack) { + responseWaiters.set(command.callerId, responseCallBack) + } + + EVENTS_SERVER.onMessage(JSON.stringify(command), thisObject.onMessage) + } catch (err) { + console.log('[ERROR] BackendServers -> Task Manager Server -> sendCommand -> Error Sending Command to Events Server -> command = ' + JSON.stringify(command)) + console.log('[ERROR] BackendServers -> Task Manager Server -> sendCommand -> Error Sending Command to Events Server -> error = ' + err.stack) + } + } + + function createEventHandler(eventHandlerName, callerId, responseCallBack) { + let eventCommand = { + action: 'createEventHandler', + eventHandlerName: eventHandlerName, + callerId: callerId + } + sendCommand(eventCommand, responseCallBack) + } + + function deleteEventHandler(eventHandlerName, callerId, responseCallBack) { + let eventCommand = { + action: 'deleteEventHandler', + eventHandlerName: eventHandlerName, + callerId: callerId + } + sendCommand(eventCommand, responseCallBack) + } + + function listenToEvent(eventHandlerName, eventType, extraData, callerId, responseCallBack, eventsCallBack) { + let eventCommand = { + action: 'listenToEvent', + eventHandlerName: eventHandlerName, + eventType: eventType, + extraData: extraData, + callerId: callerId + } + sendCommand(eventCommand, responseCallBack, eventsCallBack) + } + + function stopListening(eventHandlerName, eventType, eventSubscriptionId, callerId, responseCallBack) { + /* User either needs to specify a valid eventSubscriptionId OR the 3 params: eventHandlerName, eventType, callerId which were used when start listening to events. */ + let eventCommand = { + action: 'stopListening', + eventHandlerName: eventHandlerName, + eventType: eventType, + eventSubscriptionId: eventSubscriptionId, + callerId: callerId + } + sendCommand(eventCommand, responseCallBack) + } + + function raiseEvent(eventHandlerName, eventType, event, callerId, responseCallBack) { + let eventCommand = { + action: 'raiseEvent', + eventHandlerName: eventHandlerName, + eventType: eventType, + event: event, + callerId: callerId + } + sendCommand(eventCommand, responseCallBack) + } + } +} + + + diff --git a/webSocketsServer.js b/webSocketsServer.js new file mode 100644 index 000000000..57dbc26af --- /dev/null +++ b/webSocketsServer.js @@ -0,0 +1,112 @@ +exports.newWebSocketsServer = function newWebSocketsServer(EVENTS_SERVER) { + + const MODULE = "Web Sockets Server" + const LOG_INFO = false + + let thisObject = { + initialize: initialize, + finalize: finalize, + run: run + } + + require('dotenv').config() + const WEB_SOCKET = require('ws') + let socketServer + let port = process.env.WEB_SOCKETS_SERVER_PORT + + return thisObject + + function finalize() { + + } + + function initialize() { + + } + + function run() { + setUpWebSocketServer() + } + + function setUpWebSocketServer() { + try { + socketServer = new WEB_SOCKET.Server({ port: port }) + + socketServer.on('connection', onConnection) + + function onConnection(socket) { + + let lastNonce = -1 + if (LOG_INFO === true) { + console.log('New Websocket Connection.') + } + + socket.on('message', onMenssage) + + function onMenssage(message){ + + try { + if (LOG_INFO === true) { + console.log('Message Received: ' + message) + } + + let messageArray = message.split('|') + + let origin = messageArray[0] + let nonce = messageArray[1] + let messageToEventServer = messageArray[2] + + if (origin === 'Web Browser') { + if (isNaN(nonce) || nonce === "") { + console.log('[ERROR] BackendServers -> Web Sockets Server -> run -> setUpWebSocketServer -> Nonce is not a Number. message = ' + message) + console.log('[ERROR] BackendServers -> Web Sockets Server -> run -> setUpWebSocketServer -> Nonce is not a Number. nonce = ' + nonce) + return + } + + if (nonce === "1") { // this happens once the browser is restarted, and the server is running since a previous session. + lastNonce = 0 + } + + if (Number(nonce) < Number(lastNonce)) { + console.log('[ERROR] BackendServers -> Web Sockets Server -> run -> setUpWebSocketServer -> Nonce received is less than Last Nonce. message = ' + message) + console.log('[ERROR] BackendServers -> Web Sockets Server -> run -> setUpWebSocketServer -> Nonce received is less than Last Nonce. nonce = ' + nonce) + console.log('[ERROR] BackendServers -> Web Sockets Server -> run -> setUpWebSocketServer -> Nonce received is less than Last Nonce. lastNonce = ' + lastNonce) + return + } + + lastNonce = nonce + + try { + let jsonCheck = JSON.parse(messageToEventServer) + } catch (err) { + console.log('[ERROR] BackendServers -> Web Sockets Server -> run -> setUpWebSocketServer -> Message received from the browser is not a valid JSON. message = ' + message) + console.log('[ERROR] BackendServers -> Web Sockets Server -> run -> setUpWebSocketServer -> Message received from the browser is not a valid JSON. messageToEventServer = ' + messageToEventServer) + return + } + + let acknowledgeMessage = { + action: 'Acknowledge', + nonce: nonce + } + + socket.send(JSON.stringify(acknowledgeMessage)) + } + + + EVENTS_SERVER.onMessage(messageToEventServer, onResponse) + + function onResponse(message) { + socket.send(message) + } + } catch (err) { + console.log('[ERROR] BackendServers -> Web Sockets Server -> run -> setUpWebSocketServer -> Nonce received is less than Last Nonce. err = ' + err.stack) + } + } + } + + console.log('Web Sockets Server Started.') + } catch (err) { + console.log('[ERROR] BackendServers -> Web Sockets Server -> run -> setUpWebSocketServer -> err.message = ' + err.message) + } + } +} \ No newline at end of file From ee9441dc4dca2eea43f8386347b81a11607c2996 Mon Sep 17 00:00:00 2001 From: Luis Fernando Molina Date: Mon, 6 Apr 2020 08:17:32 +0200 Subject: [PATCH 3/5] Update eventsServer.js --- eventsServer.js | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/eventsServer.js b/eventsServer.js index 2f8fc9bb4..82730dfb7 100644 --- a/eventsServer.js +++ b/eventsServer.js @@ -107,9 +107,8 @@ if (eventHandler === undefined) { - console.log("[WARN] BackendServers -> Events Server -> onMessage -> Cannot to stop listening to events at an Event Handler that does not exist. Command = " + JSON.stringify(command)) - - sendResponse(global.DEFAULT_FAIL_RESPONSE) + /* If the event handler does not exist anymore, then there is no need to perform further actiions. */ + sendResponse(global.DEFAULT_OK_RESPONSE) return } From 1b718abb61acca6172588be33064eab8b25b6246 Mon Sep 17 00:00:00 2001 From: Luis Fernando Molina Date: Mon, 6 Apr 2020 16:13:54 +0200 Subject: [PATCH 4/5] Update webSocketsServer.js --- webSocketsServer.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/webSocketsServer.js b/webSocketsServer.js index 57dbc26af..f4737adc8 100644 --- a/webSocketsServer.js +++ b/webSocketsServer.js @@ -50,7 +50,7 @@ console.log('Message Received: ' + message) } - let messageArray = message.split('|') + let messageArray = message.split('|*|') let origin = messageArray[0] let nonce = messageArray[1] From a5a59f5ae1721c1bd016a366f859165475a4521f Mon Sep 17 00:00:00 2001 From: Luis Fernando Molina Date: Mon, 6 Apr 2020 16:34:42 +0200 Subject: [PATCH 5/5] Update eventsServer.js --- eventsServer.js | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/eventsServer.js b/eventsServer.js index 82730dfb7..cc235cf91 100644 --- a/eventsServer.js +++ b/eventsServer.js @@ -70,7 +70,7 @@ let eventHandler = eventHandlers.get(command.eventHandlerName) if (eventHandler === undefined) { /* - We will create event handlers which were not created before if a listener tries to listen to a handler that does noet + We will create event handlers which were not created before if a listener tries to listen to a handler that does not exist before the emitter comes and create it. This is to aviod syncronization problems and also problems with emitters life cycles with higher frecuency that the ones of listeners. Created in this way, event handlers are marked for deletion if all listeners stop listening at one point in time. @@ -182,7 +182,18 @@ function listenToEvent(eventHandlerName, eventType, callerId, handler, extraData) { let eventSubscriptionId = Math.trunc(Math.random() * 1000000) - thisObject.listeners.push([eventHandlerName, eventType, callerId, handler, extraData, eventSubscriptionId]) + + let found = false + for (let i = 0; i < thisObject.listeners.length; i++) { + if (eventHandlerName === thisObject.listeners[i][0] && eventType === thisObject.listeners[i][1] && callerId === thisObject.listeners[i][2]) { + found = true + } + } + + if (found === false) { + thisObject.listeners.push([eventHandlerName, eventType, callerId, handler, extraData, eventSubscriptionId]) + } + return eventSubscriptionId }