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

Removed IPC, changed it for Web Sockets

Browse Source
This commit is contained in:
Luis Fernando Molina committed 2020-04-06 08:06:47 +02:00
1 parent e07273d026
commit eb79edefde
8 files changed
+686 -293
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

TaskManager.njsproj → BackendServers.njsproj
+10 -1
View File
@@ -2,7 +2,7 @@
<PropertyGroup>
<VisualStudioVersion Condition="'$(VisualStudioVersion)' == ''">14.0</VisualStudioVersion>
<VSToolsPath Condition="'$(VSToolsPath)' == ''">$(MSBuildExtensionsPath32)\Microsoft\VisualStudio\v$(VisualStudioVersion)</VSToolsPath>
<Name>TaskManager</Name>
<Name>BackendServers</Name>
<RootNamespace>BackendServer</RootNamespace>
</PropertyGroup>
<Import Project="$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props" Condition="Exists('$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props')" />
@@ -29,7 +29,16 @@
<DebugSymbols>true</DebugSymbols>
</PropertyGroup>
<ItemGroup>
<Compile Include="eventsServer.js">
<SubType>Code</SubType>
</Compile>
<Compile Include="server.js" />
<Compile Include="taskManagerServer.js">
<SubType>Code</SubType>
</Compile>
<Compile Include="webSocketsServer.js">
<SubType>Code</SubType>
</Compile>
<Content Include="package.json" />
<Content Include="README.md" />
</ItemGroup>
TaskManager.sln → BackendServers.sln
+6 -1
View File
@@ -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
SystemEventHandler.js
-178
View File
@@ -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)
}
}
eventsServer.js
+223
View File
@@ -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)
}
}
}
}
}
package.json
+2 -2
View File
@@ -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"
}
}
server.js
+38 -111
View File
@@ -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.')
}
}
taskManagerServer.js
+295
View File
@@ -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)
}
}
}
webSocketsServer.js
+112
View File
@@ -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)
}
}
}