nope/lib/communication/mirrors/ioSocketMirrorServer.ts

/**
 * @author Martin Karkowski
 * @email m.karkowski@zema.de
 * @create date 2021-03-22 19:03:15
 * @modify date 2021-04-13 15:35:32
 * @desc [description]
 */

import * as io from "socket.io";
import { getNopeLogger } from "../../logger/getLogger";
import { LoggerLevel } from "../../logger/nopeLogger";
import { ValidEventTypesOfMirror } from "../../types/nope/nopeCommunication.interface";
import { EventMirror } from "./eventMirror";

const EVENTS_TO_FORWARD: Set<ValidEventTypesOfMirror> = new Set([
  // Default emitters
  "Aurevoir",
  "Bonjour",
  "NewInstanceGeneratorsAvailable",
  "NewInstancesAvailable",
  "NewObservablesAvailable",
  "NewServicesAvailable",
  "StatusUpdate",
  "TaskCancelation",
  "Event",
  "RpcRequest",
  "RpcResponse"
]);

/**
 * Mirror Layer using IO-Sockets.
 *
 * @export
 * @class IoSocketMirrorServer
 */
export class IoSocketMirrorServer extends EventMirror {
  protected _sockets: Set<io.Socket>;

  /**
   * Creates an instance of IoSocketMirrorServer.
   * @param {string} uri Uri of the Server.
   * @param {LoggerLevel} [level="info"] Logger level
   * @memberof IoSocketMirrorServer
   */
  constructor(public port: number, level: LoggerLevel = "info") {
    super(
      // As event Emitter, we provide the IO-Client.
      (io as any)(),
      getNopeLogger("io-mirror-server", level)
    );
    const _this = this;

    // Tell the Server to listen.
    (this._emitter as any).listen(port);

    // Now, because we arent connected we set the connected flag to false,
    // it will only be true, if a connection with this server has been established
    this.connected.getter = () => {
      return _this._sockets.size > 0;
    };

    this._logger.info("Hosting Server on Port " + port.toString());

    (this._emitter as any).on("connection", (client) => {
      _this._logger.info("New Connection established: " + client.id);

      _this._sockets.add(client);

      // Now Subscribe the Events and make shure we
      // are forwarding the data.
      for (const event of EVENTS_TO_FORWARD) {
        client.on(event, (data) => {
          if (event !== "StatusUpdate") {
            _this._logger.debug("forwarding", "'" + event.toString() + "'", data);
          }
          _this._forward(client, event, data);
        });
      }

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

        // Force an Update of the connect-flag
        _this.connected.forcePublish();
      });

      // Force an Update of the connect-flag
      _this.connected.forcePublish();
    });

    this._sockets = new Set();
  }

  /**
   * Helper Function, to forward events to the other connected Sockets.
   *
   * @protected
   * @param {io.Socket} socket The socket, which initally received the data.
   * @param {string} event the event which was received
   * @param {*} data the data, that needs to be forwarded
   * @memberof IoSocketMirrorServer
   */
  protected _forward(socket: io.Socket, event: string, data: any): void {
    // Flag, used to Debug
    let forwarded = false;
    for (const socketToForward of this._sockets) {
      if (socket !== socketToForward) {
        socketToForward.emit(event, data);
        // If data has been sended, our flag is set to true
        forwarded = true;
      }
    }

    // Now we log the output
    if (event !== "StatusUpdate") {
      this._logger.debug(forwarded ? "forwarded" : "didnt forward", "'" + event.toString() + "'", data);
    }

  }
}