595 lines
16 KiB
TypeScript
595 lines
16 KiB
TypeScript
/**
|
|
* @author Martin Karkowski
|
|
* @email m.karkowski@zema.de
|
|
* @create date 2022-01-03 21:21:45
|
|
* @modify date 2022-01-10 14:10:00
|
|
* @desc [description]
|
|
*/
|
|
|
|
import { ILogger } from "js-logger";
|
|
import { avgOfArray, maxOfArray } from "../../helpers/arrayMethods";
|
|
import { generateId } from "../../helpers/idMethods";
|
|
import { MapBasedMergeData } from "../../helpers/mergedData";
|
|
import { RUNNINGINNODE } from "../../helpers/runtimeMethods";
|
|
import { defineNopeLogger } from "../../logger/getLogger";
|
|
import { DEBUG, WARN } from "../../logger/index.browser";
|
|
import {
|
|
ENopeDispatcherStatus,
|
|
ICommunicationBridge,
|
|
IMapBasedMergeData,
|
|
INopeConnectivityManager,
|
|
INopeINopeConnectivityOptions,
|
|
INopeINopeConnectivityTimeOptions,
|
|
INopeObservable,
|
|
INopeStatusInfo,
|
|
} from "../../types/nope";
|
|
|
|
// Chached Moduls, which will be loaded in nodejs
|
|
let os = null;
|
|
let cpus = null;
|
|
|
|
// The Naviagtor for the Browser
|
|
declare const navigator;
|
|
|
|
/**
|
|
* A Modul to manage the status of other statusmanagers.
|
|
* Dispatcher should have a status manager, to ensure, the
|
|
* system is online etc.
|
|
*
|
|
* @author M.Karkowski
|
|
* @export
|
|
* @class NopeConnectivityManager
|
|
* @implements {INopeConnectivityManager}
|
|
*/
|
|
export class NopeConnectivityManager implements INopeConnectivityManager {
|
|
protected _logger: ILogger;
|
|
protected _deltaTime = 0;
|
|
protected _connectedSince: number;
|
|
|
|
/**
|
|
* The used Communication interface
|
|
*
|
|
* @type {ICommunicationBridge}
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
protected readonly _communicator: ICommunicationBridge;
|
|
|
|
/**
|
|
* A Map holding the current Status of external dispatchers.
|
|
* Key = Dispatcher-ID
|
|
* Value = Last Known status of the dispatcher
|
|
*
|
|
* @protected
|
|
* @type {Map<string, INopeStatusInfo>}
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
protected _externalDispatchers: Map<string, INopeStatusInfo>;
|
|
|
|
/**
|
|
* Timeout settings. This will define the Timers etc.
|
|
*
|
|
* @author M.Karkowski
|
|
* @protected
|
|
* @type {INopeINopeConnectivityTimeOptions}
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
protected _timeouts: INopeINopeConnectivityTimeOptions;
|
|
|
|
protected _checkInterval: any = null; // Timer to check the status
|
|
protected _sendInterval: any = null; // Timer to send the status
|
|
protected _cpuInterval: any = null; // Timer to update the CPU-Load
|
|
|
|
/**
|
|
* Internal var to hold the cpu-load
|
|
*
|
|
* @author M.Karkowski
|
|
* @protected
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
protected _cpuLoad = -1;
|
|
|
|
public readonly ready: INopeObservable<boolean>;
|
|
public readonly dispatchers: IMapBasedMergeData<
|
|
string,
|
|
string,
|
|
INopeStatusInfo
|
|
>;
|
|
|
|
/**
|
|
* Generates the current Status Message of the Dispatcher.
|
|
*
|
|
* @author M.Karkowski
|
|
* @protected
|
|
* @return {*} {IDispatcherInfo} The current status of our dispatcher.
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
public get info(): INopeStatusInfo {
|
|
if (RUNNINGINNODE) {
|
|
// If we are running our programm in node,
|
|
// we will load the corresponding libs,
|
|
// to calc the cpu load etc.
|
|
|
|
if (os === null) {
|
|
// eslint-disable-next-line
|
|
os = require("os");
|
|
}
|
|
if (cpus === null) {
|
|
// eslint-disable-next-line
|
|
cpus = os.cpus();
|
|
}
|
|
|
|
// Now lets return our status message
|
|
return {
|
|
id: this.id,
|
|
env: "javascript",
|
|
version: "1.0.0",
|
|
isMaster: this.isMaster,
|
|
host: {
|
|
cores: cpus.length,
|
|
cpu: {
|
|
model: `${cpus[0].model}`.slice(
|
|
0,
|
|
(cpus[0].model as string).indexOf("@") - 1
|
|
),
|
|
speed: avgOfArray(cpus, "speed"),
|
|
usage: this._cpuLoad,
|
|
},
|
|
os: os.platform(),
|
|
ram: {
|
|
// Return the used Memory
|
|
usedPerc: 1 - os.freemem() / os.totalmem(),
|
|
// The Values are given in Byte but we want MByte
|
|
free: Math.round(os.freemem() / 1048576),
|
|
total: Math.round(os.totalmem() / 1048576),
|
|
},
|
|
name: os.hostname(),
|
|
},
|
|
pid: process.pid,
|
|
timestamp: this.now,
|
|
upTime: this.upTime,
|
|
status: ENopeDispatcherStatus.HEALTHY,
|
|
};
|
|
}
|
|
return {
|
|
env: "javascript",
|
|
version: "1.0.0",
|
|
isMaster: this.isMaster,
|
|
host: {
|
|
cores: -1,
|
|
cpu: {
|
|
model: "unkown",
|
|
speed: -1,
|
|
usage: -1,
|
|
},
|
|
name: navigator.appCodeName + " " + navigator.appName,
|
|
os: navigator.platform,
|
|
ram: {
|
|
free: -1,
|
|
usedPerc: -1,
|
|
total: -1,
|
|
},
|
|
},
|
|
id: this.id,
|
|
pid: this.id,
|
|
timestamp: this.now,
|
|
upTime: this.upTime,
|
|
status: ENopeDispatcherStatus.HEALTHY,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Creates an instance of nopeDispatcher.
|
|
* @param {nopeRpcDispatcherOptions} options The Options, used by the Dispatcher.
|
|
* @param {() => INopeObservable<IExternalEventMsg>} _generateObservable A Helper, to generate Observables.
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
constructor(
|
|
public options: INopeINopeConnectivityOptions,
|
|
protected _generateObservable: <T>() => INopeObservable<T>,
|
|
public readonly id: string = null
|
|
) {
|
|
this._communicator = options.communicator;
|
|
|
|
this._connectedSince = Date.now();
|
|
|
|
if (id === null) {
|
|
this.id = generateId();
|
|
}
|
|
|
|
this._logger = defineNopeLogger(
|
|
options.logger,
|
|
"core.connectivity-manager"
|
|
);
|
|
|
|
// Update the Timesettings
|
|
this.setTimings(options.timeouts || {});
|
|
|
|
// Flag to show if the system is ready or not.
|
|
this.ready = this._generateObservable();
|
|
this.ready.setContent(false);
|
|
|
|
// Observable containing all Dispatcher Informations.
|
|
this._externalDispatchers = new Map();
|
|
this.dispatchers = new MapBasedMergeData(this._externalDispatchers, "id");
|
|
|
|
if (this._logger) {
|
|
this._logger.info("core.connectivity-manager", this.id, "is ready");
|
|
}
|
|
|
|
this.reset();
|
|
const _this = this;
|
|
this._init().catch((error) => {
|
|
if (_this._logger) {
|
|
_this._logger.error("Failed to intialize status manager");
|
|
_this._logger.error(error);
|
|
|
|
// Now we should exit the program (if we are running in nodejs)
|
|
if (RUNNINGINNODE) {
|
|
process.exit(1);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
// See interface description.
|
|
public get upTime(): number {
|
|
return Date.now() - this._connectedSince;
|
|
}
|
|
|
|
/**
|
|
* Internal value to store the Master.
|
|
*
|
|
* @author M.Karkowski
|
|
* @protected
|
|
* @type {boolean}
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
protected __isMaster: boolean = null;
|
|
|
|
public set isMaster(value: boolean) {
|
|
this.__isMaster = value;
|
|
}
|
|
|
|
// See interface description.
|
|
public get isMaster(): boolean {
|
|
if (this.__isMaster === null) {
|
|
const upTime = this.upTime;
|
|
for (const info of this.dispatchers.originalData.values()) {
|
|
if (info.upTime >= upTime) {
|
|
return false;
|
|
}
|
|
}
|
|
return true;
|
|
}
|
|
return this.__isMaster;
|
|
}
|
|
|
|
// See interface description.
|
|
public get master(): INopeStatusInfo {
|
|
const data = Array.from(this.dispatchers.originalData.values());
|
|
|
|
if (this.__isMaster === null) {
|
|
const idx = maxOfArray(data, "upTime").index;
|
|
return data[idx];
|
|
}
|
|
|
|
const masters = data.filter((item) => item.isMaster);
|
|
|
|
if (masters.length === 0) {
|
|
throw Error("No Master has been found !");
|
|
} else if (masters.length > 1) {
|
|
throw Error("Multiple Masters has been found!");
|
|
}
|
|
|
|
return masters[0];
|
|
}
|
|
|
|
// See interface description.
|
|
public get now(): number {
|
|
return Date.now() + this._deltaTime;
|
|
}
|
|
|
|
/**
|
|
* Internal Function, used to initialize the Dispatcher.
|
|
* It subscribes to the "Messages" of the communicator.
|
|
*
|
|
* @protected
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
protected async _init(): Promise<void> {
|
|
const _this = this;
|
|
|
|
this.ready.setContent(false);
|
|
|
|
// Now lets wait until we are ready.
|
|
// Everytime, we reconnect, we will adapt
|
|
this._communicator.connected.subscribe((connected) => {
|
|
if (connected) {
|
|
// Now we are up.
|
|
this._connectedSince = Date.now();
|
|
}
|
|
});
|
|
|
|
// Wait until the Element is connected.
|
|
await this._communicator.connected.waitFor((value) => value);
|
|
|
|
await this._communicator.on("StatusChanged", (info) => {
|
|
_this._externalDispatchers.set(info.id, info);
|
|
_this.dispatchers.update();
|
|
});
|
|
|
|
await this._communicator.on("Bonjour", ({ dispatcherId }) => {
|
|
if (_this.id !== dispatcherId) {
|
|
if (_this._logger?.enabledFor(DEBUG)) {
|
|
// If there is a Logger:
|
|
_this._logger.debug(
|
|
'Remote Dispatcher "' + dispatcherId + '" went online'
|
|
);
|
|
}
|
|
}
|
|
});
|
|
|
|
await this._communicator.on("Aurevoir", ({ dispatcherId }) => {
|
|
// Remove the Dispatcher.
|
|
_this._externalDispatchers.delete(dispatcherId);
|
|
_this.dispatchers.update();
|
|
});
|
|
|
|
if (this._logger) {
|
|
this._logger.info("core.connectivity-manager", this.id, "initialized");
|
|
}
|
|
|
|
await this.emitBonjour();
|
|
|
|
this.ready.setContent(true);
|
|
}
|
|
|
|
/**
|
|
* Function, which will be called to update the
|
|
* Status to the Dispatchers
|
|
*
|
|
* @author M.Karkowski
|
|
* @protected
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
protected _checkDispatcherHealth(): void {
|
|
const currentTime = Date.now();
|
|
let changes = false;
|
|
|
|
for (const status of this._externalDispatchers.values()) {
|
|
// determine the Difference
|
|
const diff = currentTime - status.timestamp;
|
|
|
|
// Based on the Difference Determine the Status
|
|
if (diff > this._timeouts.remove) {
|
|
// remove the Dispatcher. But be quite.
|
|
// Perhaps more dispatchers will be removed
|
|
this._removeDispatcher(status.id, true);
|
|
changes = true;
|
|
} else if (
|
|
diff > this._timeouts.dead &&
|
|
status.status !== ENopeDispatcherStatus.DEAD
|
|
) {
|
|
status.status = ENopeDispatcherStatus.DEAD;
|
|
changes = true;
|
|
} else if (
|
|
diff > this._timeouts.warn &&
|
|
diff <= this._timeouts.dead &&
|
|
status.status !== ENopeDispatcherStatus.WARNING
|
|
) {
|
|
status.status = ENopeDispatcherStatus.WARNING;
|
|
changes = true;
|
|
} else if (
|
|
diff > this._timeouts.slow &&
|
|
diff <= this._timeouts.warn &&
|
|
status.status !== ENopeDispatcherStatus.SLOW
|
|
) {
|
|
status.status = ENopeDispatcherStatus.SLOW;
|
|
changes = true;
|
|
} else if (
|
|
diff <= this._timeouts.slow &&
|
|
status.status !== ENopeDispatcherStatus.HEALTHY
|
|
) {
|
|
status.status = ENopeDispatcherStatus.HEALTHY;
|
|
changes = true;
|
|
}
|
|
}
|
|
|
|
if (changes) {
|
|
// Update the External Dispatchers
|
|
this.dispatchers.update();
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Removes a Dispatcher.
|
|
*
|
|
* @author M.Karkowski
|
|
* @protected
|
|
* @param {string} dispatcher
|
|
* @param {boolean} [quite=false]
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
protected _removeDispatcher(dispatcher: string, quite = false): void {
|
|
// Delete the Generators of the Instances.
|
|
const dispatcherInfo = this._externalDispatchers.get(dispatcher);
|
|
const deleted = this._externalDispatchers.delete(dispatcher);
|
|
|
|
if (!quite) {
|
|
this.dispatchers.update();
|
|
}
|
|
|
|
if (deleted && this._logger?.enabledFor(WARN)) {
|
|
// If there is a Logger:
|
|
this._logger.warn(
|
|
"a dispatcher on",
|
|
dispatcherInfo?.host.name || "unkown",
|
|
"went offline. ID of the Dispatcher: ",
|
|
dispatcher
|
|
);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Helper to send the current status to other statusmanagers.
|
|
*/
|
|
protected _sendStatus(): void {
|
|
this._communicator.emit("StatusChanged", this.info);
|
|
}
|
|
|
|
/**
|
|
* Helper function, which will synchronize the Timestamp.
|
|
* Timestamp must be provided in UTC (https://www.timeanddate.de/stadt/info/zeitzone/utc)
|
|
*
|
|
* @author M.Karkowski
|
|
* @param {number} timestamp The UTC-Timestamp
|
|
* @param {number} [delay=0] The Delay, since the Timestamp has been generated
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
public syncTime(timestamp: number, delay = 0) {
|
|
const _internalTimestamp = Date.now();
|
|
this._deltaTime = _internalTimestamp - timestamp - delay / 2;
|
|
}
|
|
|
|
public getStatus(id: string) {
|
|
return this._externalDispatchers.get(id);
|
|
}
|
|
|
|
/**
|
|
* Helper Function to manually emit a Bonjour!
|
|
*
|
|
* @return {*} {Promise<void>}
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
public async emitBonjour(): Promise<void> {
|
|
// Emit the Bonjour Message.
|
|
this._communicator.emit("Bonjour", { dispatcherId: this.id });
|
|
}
|
|
|
|
/**
|
|
* Function to reset the Dispatcher.
|
|
*
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
public reset(): void {
|
|
this._externalDispatchers.clear();
|
|
this.dispatchers.update(this._externalDispatchers);
|
|
}
|
|
|
|
/**
|
|
* Adapts the Timing Options and resets the internally used
|
|
* Timers etc.
|
|
*
|
|
* @author M.Karkowski
|
|
* @param {Partial<INopeINopeConnectivityTimeOptions>} options
|
|
* @memberof NopeConnectivityManager
|
|
*/
|
|
public setTimings(options: Partial<INopeINopeConnectivityTimeOptions>): void {
|
|
// Clear all Intervals etc.
|
|
this.dispose(true);
|
|
|
|
const _this = this;
|
|
|
|
this._timeouts = {
|
|
sendAliveInterval: 500,
|
|
checkInterval: 250,
|
|
slow: 1000,
|
|
warn: 2000,
|
|
dead: 5000,
|
|
remove: 10000,
|
|
};
|
|
|
|
// Define the Timeouts.
|
|
if (options) {
|
|
this._timeouts = Object.assign(this._timeouts, options);
|
|
}
|
|
|
|
if (RUNNINGINNODE) {
|
|
// eslint-disable-next-line
|
|
const os = require("os");
|
|
|
|
const getLoad = () => {
|
|
const cpus = os.cpus();
|
|
let totalTime = 0,
|
|
idleTime = 0;
|
|
|
|
// Determine the current load:
|
|
for (const cpu of cpus) {
|
|
for (const name in cpu.times) {
|
|
totalTime += cpu.times[name];
|
|
}
|
|
idleTime += cpu.times.idle;
|
|
}
|
|
|
|
return {
|
|
totalTime,
|
|
idleTime,
|
|
};
|
|
};
|
|
|
|
// Initally store the load
|
|
let oldTimes = getLoad();
|
|
this._cpuInterval = setInterval(() => {
|
|
// Get the current CPU Times.
|
|
const currentTimes = getLoad();
|
|
// Determine the difference between the old Times an the current Times.
|
|
_this._cpuLoad =
|
|
1 -
|
|
(currentTimes.idleTime - oldTimes.idleTime) /
|
|
(currentTimes.totalTime - oldTimes.totalTime);
|
|
// Store the current CPU-Times
|
|
oldTimes = currentTimes;
|
|
}, this._timeouts.sendAliveInterval);
|
|
}
|
|
|
|
// Setup Test Intervals:
|
|
if (this._timeouts.checkInterval > 0) {
|
|
// Define a Checker, which will test the status
|
|
// of the external Dispatchers.
|
|
this._checkInterval = setInterval(
|
|
() => _this._checkDispatcherHealth(),
|
|
this._timeouts.checkInterval
|
|
);
|
|
}
|
|
|
|
if (this._timeouts.sendAliveInterval > 0) {
|
|
// Define a Timer, which will emit Status updates with
|
|
// the disered delay.
|
|
this._sendInterval = setInterval(
|
|
() => _this._sendStatus(),
|
|
this._timeouts.sendAliveInterval
|
|
);
|
|
}
|
|
}
|
|
|
|
// See interface description
|
|
public getAllHosts(): string[] {
|
|
const hosts = new Set<string>();
|
|
for (const info of this.dispatchers.originalData.values()) {
|
|
hosts.add(info.host.name);
|
|
}
|
|
|
|
return Array.from(hosts);
|
|
}
|
|
|
|
/**
|
|
* Will dispose the Dispatcher. Must be called on exit for a clean exit. Otherwise it is defined as dirty exits
|
|
*/
|
|
public async dispose(quite = false): Promise<void> {
|
|
if (this._sendInterval) {
|
|
clearInterval(this._sendInterval);
|
|
}
|
|
if (this._checkInterval) {
|
|
clearInterval(this._checkInterval);
|
|
}
|
|
if (this._cpuInterval) {
|
|
clearInterval(this._cpuInterval);
|
|
}
|
|
|
|
// Emits the aurevoir Message.
|
|
if (!quite) {
|
|
this._communicator.emit("Aurevoir", { dispatcherId: this.id });
|
|
}
|
|
}
|
|
}
|