nope/lib/communication/IoSocketServer.ts

/**
 * @author Martin Karkowski
 * @email m.karkowski@zema.de
 * @create date 2020-11-06 08:52:42
 * @modify date 2020-12-03 19:03:41
 * @desc [description]
 */

import { EventEmitter } from "events";
import { Server } from "http";
import * as Logger from "js-logger";
import { ILogger } from "js-logger";
import * as io from "socket.io";
import { getNopeLogger } from "../logger/getLogger";
import { NopeObservable } from "../observables/nopeObservable";
import {
  IAvailableInstanceGeneratorsMsg,
  IAvailableInstancesMsg,
  IAvailableServicesMsg,
  IAvailableTopicsMsg,
  ICommunicationInterface,
  IExternalEventMsg,
  IRequestTaskMsg,
  IResponseTaskMsg,
  ITaskCancelationMsg
} from "../types/nope/nopeCommunication.interface";
import { INopeObservable } from "../types/nope/nopeObservable.interface";

/**
 * Wrapper Interface.
 * Creates an Interface for a Socket server
 */
export class IoSocketSeverEventEmitter {
  connected: INopeObservable<boolean>;
  protected _socket: io.Server;
  protected _sockets: Set<io.Socket>;
  protected _funcs: Map<string, (...args: any[]) => void>;
  protected _logger: ILogger;
  protected _internalEmitter: EventEmitter;

  /**
   * Creates an instance of IoSocketServer.
   * @param {number} port The Port, on which the Server should be hosted
   * @param {Server} [server] A Server shich should be used. (Otherwise a Server is created)
   * @memberof IoSocketServer
   */
  constructor(public port: number, server?: Server) {
    if (server) {
      this._socket = (io as any)(server);
    } else {
      this._socket = (io as any)();
    }

    this._internalEmitter = new EventEmitter();

    this.connected = new NopeObservable();
    this.connected.setContent(false);

    this._logger = getNopeLogger("io-socket-server");
    this._logger.info(
      "waiting for connection. Listening on port: " + port.toString()
    );

    this._socket.listen(port);

    const _this = this;
    this._socket.on("connection", (client) => {
      _this._logger.debug("New Connection established: " + client.id);

      // Add the Elements to the Client
      _this._sockets.add(client);

      // Subscribe to Loosing connection:
      client.on("disconnect", () => {
        _this._logger.debug("Connection of : " + client.id + " lost.");
        _this._sockets.delete(client);

        if (_this._sockets.size === 0) {
          _this.connected.setContent(false);
          _this._logger.warn("All Connections lost");
        }
      });

      for (const [id, _func] of _this._funcs) {
        // Subscribe to the Events:
        client.on(id, (...args) => {
          _this._logger.debug("received content: ", ...args);

          // Forward the Message.
          _func(...args);
        });
      }

      _this.connected.setContent(true);
    });

    this._sockets = new Set();
    this._funcs = new Map<string, (...args: any[]) => void>();
  }
  off(event: string, cb: (...args: any[]) => void): void {
    // Remove the Subscription from the Socket.
    for (const socket of this._sockets) {
      socket.off(event, cb);
    }

    // Remove the Function.
    this._funcs.delete(event);
  }

  on(event: string, cb: (...args: any[]) => void): void {
    // Add the Subscription to the Sockets.
    const _this = this;
    for (const socket of this._sockets) {
      socket.on(event, cb);
      socket.on(event, (...args) => {
        _this._socket.emit(event, ...args);
      });
    }

    // Store the Function
    this._funcs.set(event, cb);
    this._internalEmitter.on(event, cb);
  }

  emit(event: string, data: any): void {
    if (this._logger.enabledFor(Logger.DEBUG)) {
      this._logger.debug("sending data on: " + event);
    }
    this._socket.emit(event, data);
    this._internalEmitter.emit(event, data);
  }
}

/**
 * Communication Layer using IO-Sockets.
 *
 * @export
 * @class IoSocketServer
 * @implements {ICommunicationInterface}
 */
export class IoSocketServer
  extends IoSocketSeverEventEmitter
  implements ICommunicationInterface {
  subscriptionMode?: "individual" | "generic";
  resultSharing?: "individual" | "generic";
  async onTaskCancelation(
    cb: (msg: ITaskCancelationMsg) => void
  ): Promise<void> {
    this.on("cancel", cb);
  }

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

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

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

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

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

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

  async emitRpcResult(name: string, result: IResponseTaskMsg): Promise<void> {
    this.emit(name, result);
  }

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

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

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

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

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

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

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

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

  async emitNewObersvablesAvailable(
    topics: IAvailableTopicsMsg
  ): Promise<void> {
    this.emit("topics", topics);
  }

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

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

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

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

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

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