From eb79edefdef28e256551952306de902bf07c5e8d Mon Sep 17 00:00:00 2001 From: Luis Fernando Molina Date: Mon, 6 Apr 2020 08:06:47 +0200 Subject: [PATCH] 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