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..cc235cf91
--- /dev/null
+++ b/eventsServer.js
@@ -0,0 +1,233 @@
+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 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.
+ */
+ 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) {
+
+ /* If the event handler does not exist anymore, then there is no need to perform further actiions. */
+ sendResponse(global.DEFAULT_OK_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)
+
+ 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
+ }
+
+ 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 d426bd320..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)
- })
-
- setTimeout(sendStartingEvent, 3000)
-
- 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..f4737adc8
--- /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