nope/lib/communication/eventLayer.ts

/**
 * @author Martin Karkowski
 * @email m.karkowski@zema.de
 * @create date 2020-11-06 08:52:36
 * @modify date 2020-11-11 16:56:24
 * @desc [description]
 */

import { EventEmitter } from "events";
import { Logger } from "winston";
import { NopeObservable } from "../observables/nopeObservable";
import { IAvailableInstanceGeneratorsMsg, IAvailableServicesMsg, IAvailableTopicsMsg, ICommunicationInterface, IExternalEventMsg, IRequestTaskMsg, IResponseTaskMsg, ITaskCancelationMsg } from "../types/nope/nopeCommunication.interface";
import { INopeObservable } from "../types/nope/nopeObservable.interface";

/**
 * A Communication Layer for the Dispatchers.
 * Here, only a Events are used.
 *
 * @export
 * @class EventLayer
 * @implements {ICommunicationInterface}
 */
export class EventLayer implements ICommunicationInterface {
  connected: INopeObservable<boolean>;

  protected _emitter = new EventEmitter();

  constructor(
    public readonly subscriptionMode: "individual" | "generic" = "individual",
    public readonly resultSharing: "individual" | "generic" = "individual",
    protected _logger?: Logger
  ) {
    this.connected = new NopeObservable();
    this.connected.setContent(true);
  }

  async onTaskCancelation(cb: (msg: ITaskCancelationMsg) => void) {
    this._emitter.on("cancel", cb);
  }

  async emitTaskCancelation(msg: ITaskCancelationMsg) {
    this._emitter.emit("cancel", msg);
  }

  async onAurevoir(cb: (dispatcher: string) => void) {
    this._emitter.on("aurevoir", cb);
  }

  async emitAurevoir(dispatcher: string) {
    this._emitter.emit("aurevoir", dispatcher);
  }

  async emitNewInstanceGeneratorsAvailable(generators: IAvailableInstanceGeneratorsMsg) {
    this._emitter.emit("generators", generators);
  }

  async onNewInstanceGeneratorsAvailable(cb: (generators: IAvailableInstanceGeneratorsMsg) => void) {
    this._emitter.on("generators", cb);
  }

  async emitRpcRequest(name: string, request: IRequestTaskMsg) {
    this._emitter.emit(name, request);
  }

  async emitRpcResult(name: string, result: IResponseTaskMsg) {
    this._emitter.emit(name, result);
  }

  async onRpcResult(name: string, cb: (result: IResponseTaskMsg) => void) {
    this._emitter.on(name, cb);
  }

  async offRpcResponse(name: string, cb: (result: IResponseTaskMsg) => void) {
    this._emitter.off(name, cb);
  }

  async onRpcRequest(name: string, cb: (data: IRequestTaskMsg) => void) {
    this._emitter.on(name, cb);
  }

  async offRpcRequest(name: string, cb: (data: IRequestTaskMsg) => void) {
    this._emitter.off(name, cb);
  }

  async emitNewServicesAvailable(services: IAvailableServicesMsg) {
    this._emitter.emit("services", services);
  }

  async onNewServicesAvailable(cb: (services: IAvailableServicesMsg) => void) {
    this._emitter.on("services", cb);
  }

  async onBonjour(cb: (dispatcher: string) => void) {
    this._emitter.on("bonjour", cb);
  }

  async emitBonjour(dispatcher: string) {
    this._emitter.emit("bonjour", dispatcher);
  }

  async emitNewTopicsAvailable(topics: IAvailableTopicsMsg) {
    this._emitter.emit("topics", topics);
  }

  async onNewTopicsAvailable(cb: (topics: IAvailableTopicsMsg) => void) {
    this._emitter.on("topics", cb);
  }

  async onEvent(event: string, cb: (data: IExternalEventMsg) => void) {
    this._emitter.on("event_" + event, cb);
  }

  async emitEvent(event: string, data: IExternalEventMsg) {
    this._emitter.emit("event_" + event, data);
  }

  async offEvent(event: string, cb: (data: IExternalEventMsg) => void) {
    this._emitter.off("event_" + event, cb);
  }
}