217 lines
6.5 KiB
TypeScript
217 lines
6.5 KiB
TypeScript
/**
|
|
* @author Martin Karkowski
|
|
* @email m.karkowski@zema.de
|
|
* @create date 2020-11-06 08:52:36
|
|
* @modify date 2020-12-04 17:44:34
|
|
* @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",
|
|
"emitRpcResult",
|
|
"emitTaskCancelation",
|
|
"onAurevoir",
|
|
"onBonjour",
|
|
"onNewInstanceGeneratorsAvailable",
|
|
"onNewInstancesAvailable",
|
|
"onNewObservablesAvailable",
|
|
"onNewServicesAvailable",
|
|
"onTaskCancelation",
|
|
];
|
|
|
|
const STORING: Array<keyof ICommunicationInterface> = [
|
|
"onAurevoir",
|
|
"onBonjour",
|
|
"onNewInstanceGeneratorsAvailable",
|
|
"onNewInstancesAvailable",
|
|
"onNewObservablesAvailable",
|
|
"onNewServicesAvailable",
|
|
"onTaskCancelation",
|
|
];
|
|
|
|
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: Set<ICommunicationInterface>;
|
|
|
|
public addLayer(layer: ICommunicationInterface) {
|
|
if (!this._layers.has(layer)) {
|
|
this._layers.add(layer);
|
|
|
|
if (this._enableDynamicAdding) {
|
|
// Play the history:
|
|
for (const method of STORING) {
|
|
for (const args of this[`_map_${method}`]) {
|
|
layer[method as any](...args);
|
|
}
|
|
}
|
|
|
|
// 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];
|
|
// 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[dict[container]](event,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 Set<ICommunicationInterface>();
|
|
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){
|
|
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 (...args) => {
|
|
// Store the Call
|
|
_this[`_map_${method}`].push(args);
|
|
|
|
for (const _layer of _this._layers) {
|
|
_layer[method as any](...args);
|
|
}
|
|
};
|
|
} else if
|
|
// Else if the method store the events individually =>
|
|
// Creaete a different Method:
|
|
(_enableDynamicAdding &&
|
|
(SPECIFIC_STORING.includes(method) || SPECIFIC_REMOVING.includes(method))) {
|
|
// Determine the Container name.
|
|
const _container = SPECIFIC_STORING.includes(method) ? `_${method}` : method.replace("off","_on");
|
|
// Define the Container itself
|
|
this[_container] = new Map<string, Set<any>>();
|
|
// Based on the Function determine the Action:
|
|
const mode: "add" | "delete" = SPECIFIC_STORING.includes(method) ? "add": "delete";
|
|
// Store the Method.
|
|
this[method as any] = async (name, cb) => {
|
|
|
|
// Call the adapted Function
|
|
_this._adaptStore(_container as any,mode, name, cb);
|
|
|
|
// Perform the Action on every available Layer.
|
|
for (const _layer of _this._layers) {
|
|
_layer[method as any](name, 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) {
|
|
_layer[method as any](...args);
|
|
}
|
|
};
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 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);
|
|
}
|
|
}
|
|
}
|