/** * @author Martin Karkowski * @email m.karkowski@zema.de * @create date 2020-11-06 08:52:36 * @modify date 2020-12-30 15:09:24 * @desc [description] */ import { EventEmitter } from "events"; import { ILogger } from "js-logger"; import { NopeObservable } from "../observables/nopeObservable"; import { IAvailableInstanceGeneratorsMsg, IAvailableInstancesMsg, IAvailableServicesMsg, IAvailableTopicsMsg, ICommunicationInterface, IExternalEventMsg, IRequestTaskMsg, IResponseTaskMsg, ITaskCancelationMsg } from "../types/nope/nopeCommunication.interface"; import { IDispatcherInfo } from "../types/nope/nopeDispatcher.interface"; import { INopeObservable } from "../types/nope/nopeObservable.interface"; export interface IEmitter { on(event: string | symbol, listener: (...args: any[]) => void): this; off(event: string | symbol, listener: (...args: any[]) => void): this; emit(event: string | symbol, ...args: any[]): boolean; getMaxListeners?: () => number; setMaxListeners?: (n: number) => void; } /** * A Communication Layer for the Dispatchers. * Here, only a Events are used. * * @export * @class EventLayer * @implements {ICommunicationInterface} */ export class EventLayer implements ICommunicationInterface { connected: INopeObservable; /** * Creaetes an Event Emitter-based Communication Layer: * @param _emitter * @param subscriptionMode * @param resultSharing * @param _logger */ constructor( protected _emitter: IEmitter = new EventEmitter(), public readonly subscriptionMode: "individual" | "generic" = "individual", public readonly resultSharing: "individual" | "generic" = "individual", protected _logger?: ILogger ) { this.connected = new NopeObservable(); this.connected.setContent(true); } protected _on(event: string, cb) { if (typeof this._emitter.getMaxListeners === "function") { this._emitter.setMaxListeners(this._emitter.getMaxListeners() + 1); } this._emitter.on(event, cb); } async emitStatusUpdate(status: IDispatcherInfo): Promise { this._emitter.emit("statusUdpate", status); } async onStatusUpdate(cb: (status: IDispatcherInfo) => void): Promise { this._on("statusUdpate", cb); } async onNewInstancesAvailable( cb: (instances: IAvailableInstancesMsg) => void ): Promise { this._on("newInstancesAvailable", cb); } async emitNewInstancesAvailable( instances: IAvailableInstancesMsg ): Promise { this._emitter.emit("newInstancesAvailable", instances); } async onTaskCancelation( cb: (msg: ITaskCancelationMsg) => void ): Promise { this._on("cancel", cb); } async emitTaskCancelation(msg: ITaskCancelationMsg): Promise { this._emitter.emit("cancel", msg); } async onAurevoir(cb: (dispatcher: string) => void): Promise { this._on("aurevoir", cb); } async emitAurevoir(dispatcher: string): Promise { this._emitter.emit("aurevoir", dispatcher); } async emitNewInstanceGeneratorsAvailable( generators: IAvailableInstanceGeneratorsMsg ): Promise { this._emitter.emit("generators", generators); } async onNewInstanceGeneratorsAvailable( cb: (generators: IAvailableInstanceGeneratorsMsg) => void ): Promise { this._on("generators", cb); } async emitRpcRequest(name: string, request: IRequestTaskMsg): Promise { this._emitter.emit(name, request); } async emitRpcResponse(name: string, result: IResponseTaskMsg): Promise { this._emitter.emit(name, result); } async onRpcResponse( name: string, cb: (result: IResponseTaskMsg) => void ): Promise { this._on(name, cb); } async offRpcResponse( name: string, cb: (result: IResponseTaskMsg) => void ): Promise { this._emitter.off(name, cb); } async onRpcRequest( name: string, cb: (data: IRequestTaskMsg) => void ): Promise { this._on(name, cb); } async offRpcRequest( name: string, cb: (data: IRequestTaskMsg) => void ): Promise { this._emitter.off(name, cb); } async emitNewServicesAvailable( services: IAvailableServicesMsg ): Promise { this._emitter.emit("services", services); } async onNewServicesAvailable( cb: (services: IAvailableServicesMsg) => void ): Promise { this._on("services", cb); } async onBonjour(cb): Promise { this._on("bonjour", cb); } async emitBonjour(msg): Promise { this._emitter.emit("bonjour", msg); } async emitNewObersvablesAvailable( topics: IAvailableTopicsMsg ): Promise { this._emitter.emit("topics", topics); } async onNewObservablesAvailable( cb: (topics: IAvailableTopicsMsg) => void ): Promise { this._on("topics", cb); } async onEvent( event: string, cb: (data: IExternalEventMsg) => void ): Promise { this._on("event_" + event, cb); } async emitEvent(event: string, data: IExternalEventMsg): Promise { this._emitter.emit("event_" + event, data); } async offEvent( event: string, cb: (data: IExternalEventMsg) => void ): Promise { this._emitter.off("event_" + event, cb); } }