320 lines
9.6 KiB
TypeScript
320 lines
9.6 KiB
TypeScript
/**
|
|
* @author Martin Karkowski
|
|
* @email m.karkowski@zema.de
|
|
* @create date 2020-11-06 08:52:36
|
|
* @modify date 2021-01-12 08:54:17
|
|
* @desc [description]
|
|
*/
|
|
|
|
import { ILogger } from "js-logger";
|
|
import { NopeObservable } from "../observables/nopeObservable";
|
|
import { ICommunicationInterface } from "../types/nope/nopeCommunication.interface";
|
|
import { INopeObservable } from "../types/nope/nopeObservable.interface";
|
|
|
|
const METHODS: Array<keyof ICommunicationInterface> = [
|
|
"emitAurevoir",
|
|
"emitBonjour",
|
|
"emitEvent",
|
|
"emitNewInstanceGeneratorsAvailable",
|
|
"emitNewInstancesAvailable",
|
|
"emitNewObersvablesAvailable",
|
|
"emitNewServicesAvailable",
|
|
"emitRpcRequest",
|
|
"emitRpcResponse",
|
|
"emitStatusUpdate",
|
|
"emitTaskCancelation",
|
|
"onAurevoir",
|
|
"onBonjour",
|
|
"onNewInstanceGeneratorsAvailable",
|
|
"onNewInstancesAvailable",
|
|
"onNewObservablesAvailable",
|
|
"onNewServicesAvailable",
|
|
"onStatusUpdate",
|
|
"onTaskCancelation"
|
|
];
|
|
|
|
const METHOD_MAPPING: {
|
|
[P in keyof ICommunicationInterface]: keyof ICommunicationInterface;
|
|
} = {
|
|
onAurevoir: "emitAurevoir",
|
|
onBonjour: "emitBonjour",
|
|
onEvent: "emitEvent",
|
|
onNewInstanceGeneratorsAvailable: "emitNewInstanceGeneratorsAvailable",
|
|
onNewInstancesAvailable: "emitNewInstancesAvailable",
|
|
onNewObservablesAvailable: "emitNewObersvablesAvailable",
|
|
onNewServicesAvailable: "emitNewServicesAvailable",
|
|
onRpcRequest: "emitRpcRequest",
|
|
onRpcResponse: "emitRpcResponse",
|
|
onTaskCancelation: "emitTaskCancelation",
|
|
onStatusUpdate: "emitStatusUpdate"
|
|
};
|
|
|
|
const STORING: Array<keyof ICommunicationInterface> = [
|
|
"onAurevoir",
|
|
"onBonjour",
|
|
"onNewInstanceGeneratorsAvailable",
|
|
"onNewInstancesAvailable",
|
|
"onNewObservablesAvailable",
|
|
"onNewServicesAvailable",
|
|
"onTaskCancelation",
|
|
"onStatusUpdate"
|
|
];
|
|
|
|
const SPECIFIC_STORING: Array<keyof ICommunicationInterface> = [
|
|
"onRpcResponse",
|
|
"onRpcRequest",
|
|
"onEvent"
|
|
];
|
|
|
|
const SPECIFIC_REMOVING: Array<keyof ICommunicationInterface> = [
|
|
"offEvent",
|
|
"offRpcRequest",
|
|
"offRpcResponse"
|
|
];
|
|
|
|
METHODS.push(...SPECIFIC_REMOVING, ...SPECIFIC_STORING);
|
|
|
|
/**
|
|
* A Communication Layer for the Dispatchers.
|
|
* Here, only a Events are used.
|
|
*
|
|
* @export
|
|
* @class EventLayer
|
|
* @implements {ICommunicationInterface}
|
|
*/
|
|
//@ts-ignore Ignore the Interface. Its implemented manually
|
|
export class Bridge implements ICommunicationInterface {
|
|
connected: INopeObservable<boolean>;
|
|
|
|
protected _layers: Map<ICommunicationInterface, boolean>;
|
|
|
|
public addLayer(layer: ICommunicationInterface, forward = false) {
|
|
if (!this._layers.has(layer)) {
|
|
this._layers.set(layer, forward);
|
|
|
|
if (this._enableDynamicAdding) {
|
|
const _this = this;
|
|
|
|
// Play the history:
|
|
for (const method of STORING) {
|
|
for (const cb of this[`_map_${method}`]) {
|
|
layer[method as any](
|
|
forward ? this._wrapMethodWithoutName(cb, layer, method) : cb
|
|
);
|
|
}
|
|
}
|
|
|
|
// Create a dictionary, which maps the container
|
|
// to the corresponding Method.
|
|
const dict: {
|
|
[index: string]: "onRpcResponse" | "onRpcRequest" | "onEvent";
|
|
} = {
|
|
_onRpcResponse: "onRpcResponse",
|
|
_onRpcRequest: "onRpcRequest",
|
|
_onEvent: "onEvent"
|
|
};
|
|
|
|
// Now create a Loop that performs the adding of
|
|
// adding all subscribed responses, request etc.
|
|
for (const container in dict) {
|
|
// 1. Extract the element:
|
|
const map: Map<string, Set<any>> = this[container];
|
|
const method = dict[container];
|
|
// 2. Iterate over the Elements.
|
|
for (const [event, callbacks] of map.entries()) {
|
|
// 3. Add all callbacks (Use the Map operator.)
|
|
Array.from(callbacks).map((cb) =>
|
|
layer[method](
|
|
event,
|
|
forward ? _this._wrapMethod(event, cb, layer, method) : cb
|
|
)
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
public removeLayer(layer: ICommunicationInterface) {
|
|
if (this._layers.has(layer)) {
|
|
// Delete the Layer
|
|
this._layers.delete(layer);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Creates a Bridge
|
|
* @param subscriptionMode
|
|
* @param resultSharing
|
|
* @param _enableDynamicAdding
|
|
* @param _logger
|
|
*/
|
|
constructor(
|
|
public readonly subscriptionMode: "individual" | "generic" = "individual",
|
|
public readonly resultSharing: "individual" | "generic" = "individual",
|
|
protected _enableDynamicAdding = false,
|
|
protected _logger?: ILogger
|
|
) {
|
|
this._layers = new Map();
|
|
this.connected = new NopeObservable();
|
|
this.connected.setContent(false);
|
|
|
|
const _this = this;
|
|
|
|
// Add a custom handler for the connect flag.
|
|
// the Flag is defined as true, if every socket
|
|
// is connected.
|
|
this.connected.getter = () => {
|
|
for (const layer of _this._layers.keys()) {
|
|
if (!layer.connected.getContent()) return false;
|
|
}
|
|
|
|
return true;
|
|
};
|
|
|
|
// Iterate over the Methods of the Element.
|
|
// Define for every method the corresponding method.
|
|
for (const method of METHODS) {
|
|
// If the Subscription should be stored and dynamic Adding is enabled =>
|
|
// Add the methods.
|
|
if (_enableDynamicAdding && STORING.includes(method)) {
|
|
this[`_map_${method}`] = new Array<any[]>();
|
|
// Define a Function which stores the Actions:
|
|
this[method as any] = async (cb) => {
|
|
// Store the Call
|
|
_this[`_map_${method}`].push(cb);
|
|
|
|
for (const [_layer, _forward] of _this._layers) {
|
|
_layer[method as any](
|
|
_forward ? _this._wrapMethodWithoutName(cb, _layer, method) : cb
|
|
);
|
|
}
|
|
};
|
|
} else if (
|
|
// Else if the method store the events individually =>
|
|
// Creaete a different Method:
|
|
_enableDynamicAdding &&
|
|
SPECIFIC_STORING.includes(method)
|
|
) {
|
|
// Determine the Container name.
|
|
const _container = `_${method}`;
|
|
// Define the Container itself
|
|
this[_container] = new Map<string, Set<any>>();
|
|
|
|
// Store the Method.
|
|
this[method as any] = async (name, cb) => {
|
|
// Call the adapted Function
|
|
_this._adaptStore(_container as any, "add", name, cb);
|
|
|
|
// Perform the Action on every available Layer.
|
|
for (const [_layer, _forward] of _this._layers) {
|
|
// Store the Callback. Because it is a "method" which listens on the
|
|
// events => wrap the method, to forward the event to the other layers.
|
|
_layer[method as any](
|
|
name,
|
|
_forward ? _this._wrapMethod(name, cb, _layer, method) : cb
|
|
);
|
|
}
|
|
};
|
|
} else if (
|
|
// Else if the method store the events individually =>
|
|
// Creaete a different Method:
|
|
_enableDynamicAdding &&
|
|
SPECIFIC_REMOVING.includes(method)
|
|
) {
|
|
// Determine the Container name.
|
|
const _container = method.replace("off", "_on");
|
|
// Define the Container itself
|
|
this[_container] = new Map<string, Set<any>>();
|
|
// Store the Method.
|
|
this[method as any] = async (name, cb) => {
|
|
// Call the adapted Function
|
|
_this._adaptStore(_container as any, "delete", name, cb);
|
|
|
|
// Perform the Action on every available Layer.
|
|
for (const [_layer, _forward] of _this._layers) {
|
|
// Store the Callback
|
|
_layer[method as any](name, _this._wrappedMethods.get(cb));
|
|
}
|
|
|
|
// Delete the Wrapped Method:
|
|
_this._wrappedMethods.delete(cb);
|
|
};
|
|
} else {
|
|
// Define a Function which stores the Actions:
|
|
this[method as any] = async (...args) => {
|
|
if (_this._logger) {
|
|
_this._logger.debug(method, ...args);
|
|
}
|
|
|
|
for (const _layer of _this._layers.keys()) {
|
|
_layer[method as any](...args);
|
|
}
|
|
};
|
|
}
|
|
}
|
|
}
|
|
|
|
protected _wrappedMethods = new Map<(data) => any, (data) => any>();
|
|
protected _wrapMethod(
|
|
name: string,
|
|
cb: (data) => any,
|
|
layer: ICommunicationInterface,
|
|
method: keyof ICommunicationInterface
|
|
): (data) => any {
|
|
const _this = this;
|
|
const _wrapped = (data) => {
|
|
// _this._logger.info("Forwarding",method,name);
|
|
for (const _layer of _this._layers.keys()) {
|
|
if (_layer != layer) {
|
|
// Emit the corresponding Event on the different channels.
|
|
_layer[METHOD_MAPPING[method] as any](name, data);
|
|
}
|
|
}
|
|
// Call the Callback!
|
|
cb(data);
|
|
};
|
|
|
|
this._wrappedMethods.set(cb, _wrapped);
|
|
return _wrapped;
|
|
}
|
|
protected _wrapMethodWithoutName(
|
|
cb: (data) => any,
|
|
layer: ICommunicationInterface,
|
|
method: keyof ICommunicationInterface
|
|
): (data) => any {
|
|
const _this = this;
|
|
return (data) => {
|
|
// _this._logger.info("Forwarding",method);
|
|
for (const _layer of _this._layers.keys()) {
|
|
if (_layer != layer) {
|
|
// Emit the corresponding Event on the different channels.
|
|
_layer[METHOD_MAPPING[method] as any](data);
|
|
}
|
|
}
|
|
// Call the Callback!
|
|
cb(data);
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Helper Function that is used to adapt the current active subscriptions
|
|
* @param name name of the data container
|
|
* @param mode the mode (delete or add)
|
|
* @param event the name of the event
|
|
* @param cb the callback
|
|
*/
|
|
protected _adaptStore(
|
|
name: "_onRpcResponse" | "_onRpcRequest" | "_onEvent",
|
|
mode: "add" | "delete",
|
|
event: string,
|
|
cb: any
|
|
): void {
|
|
if (this._enableDynamicAdding) {
|
|
const _set01 = this[name].get(event) || new Set();
|
|
_set01[mode](cb);
|
|
this[name].set(event, _set01);
|
|
}
|
|
}
|
|
}
|