nope/lib/communication/layers/eventLayer.ts

/**
 * @author Martin Karkowski
 * @email m.karkowski@zema.de
 * @create date 2020-11-06 08:52:36
 * @modify date 2021-04-09 08:36:53
 * @desc [description]
 */

import { EventEmitter } from "events";
import { ILogger } from "js-logger";
import { generateId } from "../../helpers/idMethods";
import { NopeObservable } from "../../observables/nopeObservable";
import {
  IAvailableInstanceGeneratorsMsg,
  IAvailableInstancesMsg,
  IAvailableServicesMsg,
  IAvailableTopicsMsg,
  ICommunicationInterface,
  IEmitter,
  IExecutingTaskMsg,
  IExternalEventMsg,
  IRequestTaskMsg,
  IResponseTaskMsg,
  IRpcUnregisterMsg,
  ITaskCancelationMsg
} from "../../types/nope/nopeCommunication.interface";
import { IDispatcherInfo } from "../../types/nope/nopeDispatcher.interface";
import { INopeObservable } from "../../types/nope/nopeObservable.interface";

/**
 * A Communication Layer for the Dispatchers.
 * Here, only a Events are used. This layer represents
 * the basic layer. If transmits the events using
 * internal emitters.
 *
 * @export
 * @class EventLayer
 * @implements {ICommunicationInterface}
 */
export class EventLayer implements ICommunicationInterface {
  /**
   * Flag, showing whether the Mirror is connected or not.
   *
   * @type {INopeObservable<boolean>}
   * @memberof EventLayer
   */
  connected: INopeObservable<boolean>;
  /**
   * Flag, to enable, that this layer will be used in a bridge,
   * to show, whether the bridge is connected or not.
   *
   * @memberof EventLayer
   */
  considerConnection = true;

  allowServiceRedundancy = false;

  /**
   * Creaetes an Event Emitter-based Communication Layer:
   * @param _emitter
   * @param _logger
   */
  constructor(
    protected _emitter: IEmitter = new EventEmitter(),
    protected _logger?: ILogger
  ) {
    this.connected = new NopeObservable();
    this.connected.setContent(true);
    this.id = generateId();
    this._subscribing = new Map();
  }

  id: string;

  protected _subscribing: Map<string, Set<(...args) => any>>;

  /**
   * Helper-Function, which will publish the subescribe events
   *
   * @protected
   * @memberof EventLayer
   */
  protected _publishSubscriptions(): void {
    const activeSubscriptions = Array.from(this._subscribing.entries())
      .filter((entry) => {
        return entry[1].size > 0;
      })
      .map((entry) => entry[0]);

    this._emitter.emit("subscribing", activeSubscriptions);
  }

  protected async _on(event: string, cb): Promise<void> {
    if (typeof (await this._emitter.getMaxListeners) === "function") {
      await this._emitter.setMaxListeners(
        (await this._emitter.getMaxListeners()) + 1
      );
    }

    // Store the Event we are publishing
    if (!this._subscribing.has(event)) {
      this._subscribing.set(event, new Set());
    }
    this._subscribing.get(event).add(cb);

    this._publishSubscriptions();
    this._emitter.on(event, cb);
  }

  protected async _emit(name: string, data: any): Promise<void> {
    this._emitter.emit("emitted", { name, data });
  }

  protected async _off(name, cb): Promise<void> {
    this._off(name, cb);

    // Store the Event we are publishing
    if (!this._subscribing.has(name)) {
      this._subscribing.set(name, new Set());
    }
    this._subscribing.get(name).delete(cb);

    this._publishSubscriptions();
  }

  async emitStatusUpdate(status: IDispatcherInfo): Promise<void> {
    await this._emit("statusUdpate", status);
  }

  async onStatusUpdate(cb: (status: IDispatcherInfo) => void): Promise<void> {
    await this._on("statusUdpate", cb);
  }

  async onNewInstancesAvailable(
    cb: (instances: IAvailableInstancesMsg) => void
  ): Promise<void> {
    await this._on("newInstancesAvailable", cb);
  }

  async emitNewInstancesAvailable(
    instances: IAvailableInstancesMsg
  ): Promise<void> {
    await this._emit("newInstancesAvailable", instances);
  }

  async onTaskCancelation(
    cb: (msg: ITaskCancelationMsg) => void
  ): Promise<void> {
    await this._on("cancel", cb);
  }

  async emitTaskCancelation(msg: ITaskCancelationMsg): Promise<void> {
    await this._emit("cancel", msg);
  }

  async onAurevoir(cb: (dispatcher: string) => void): Promise<void> {
    await this._on("aurevoir", cb);
  }

  async emitAurevoir(dispatcher: string): Promise<void> {
    await this._emit("aurevoir", dispatcher);
  }

  async emitNewInstanceGeneratorsAvailable(
    generators: IAvailableInstanceGeneratorsMsg
  ): Promise<void> {
    await this._emit("generators", generators);
  }

  async onNewInstanceGeneratorsAvailable(
    cb: (generators: IAvailableInstanceGeneratorsMsg) => void
  ): Promise<void> {
    await this._on("generators", cb);
  }

  async emitRpcRequest(name: string, request: IRequestTaskMsg): Promise<void> {
    await this._emit(name, request);
  }

  async emitRpcResponse(name: string, result: IResponseTaskMsg): Promise<void> {
    await this._emit(name, result);
  }

  async onRpcResponse(
    name: string,
    cb: (result: IResponseTaskMsg) => void
  ): Promise<void> {
    await this._on(name, cb);
  }

  async offRpcResponse(
    name: string,
    cb: (result: IResponseTaskMsg) => void
  ): Promise<void> {
    await this._off(name, cb);
  }

  async onRpcRequest(
    name: string,
    cb: (data: IRequestTaskMsg) => void
  ): Promise<void> {
    await this._on(name, cb);
  }

  async offRpcRequest(
    name: string,
    cb: (data: IRequestTaskMsg) => void
  ): Promise<void> {
    await this._off(name, cb);
  }

  async emitNewServicesAvailable(
    services: IAvailableServicesMsg
  ): Promise<void> {
    await this._emit("services", services);
  }

  async onNewServicesAvailable(
    cb: (services: IAvailableServicesMsg) => void
  ): Promise<void> {
    await this._on("services", cb);
  }

  async onBonjour(cb): Promise<void> {
    await this._on("bonjour", cb);
  }

  async emitBonjour(msg): Promise<void> {
    await this._emit("bonjour", msg);
  }

  async emitNewObservablesAvailable(
    topics: IAvailableTopicsMsg
  ): Promise<void> {
    await this._emit("topics", topics);
  }

  async onNewObservablesAvailable(
    cb: (topics: IAvailableTopicsMsg) => void
  ): Promise<void> {
    await this._on("topics", cb);
  }

  async onEvent(
    event: string,
    cb: (data: IExternalEventMsg) => void
  ): Promise<void> {
    await this._on("event_" + event, cb);
  }

  async emitEvent(event: string, data: IExternalEventMsg): Promise<void> {
    await this._emit("event_" + event, data);
  }

  async offEvent(
    event: string,
    cb: (data: IExternalEventMsg) => void
  ): Promise<void> {
    await this._off("event_" + event, cb);
  }

  async onUnregisterRpc(cb: (msg: IRpcUnregisterMsg) => void): Promise<void> {
    await this._on("RpcUnregister", cb);
  }

  async emitUnregisterRpc(data: IRpcUnregisterMsg): Promise<void> {
    await this._emit("RpcUnregister", data);
  }

  async onExecutingTasks(cb: (msg: IExecutingTaskMsg) => void): Promise<void> {
    await this._on("ExecutingTask", cb);
  }

  async emitExecutingTasks(data: IExecutingTaskMsg): Promise<void> {
    await this._emit("ExecutingTask", data);
  }

  async dispose(): Promise<void> {
    this.connected.dispose();
  }
}