nope/lib/communication/IoSocketClient.ts

/**
 * @author Martin Karkowski
 * @email m.karkowski@zema.de
 * @create date 2020-11-04 17:36:04
 * @modify date 2020-12-04 17:11:18
 * @desc [description]
 */

import { EventEmitter } from "events";
import { ILogger } from "js-logger";
import { connect, Socket } from "socket.io-client";
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";

export class IoSocketClient implements ICommunicationInterface {
  connected: INopeObservable<boolean>;

  protected _emitter: typeof Socket;
  protected _internalEmitter: EventEmitter;
  protected _logger: ILogger;

  constructor(public uri: string) {
    const _this = this;

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

    // Make shure we use the http before connecting.
    this.uri = this.uri.startsWith("http://") ? this.uri : "http://" + this.uri;

    // Create a Logger
    this._logger = getNopeLogger("io-socket-client");
    this._logger.info("connecting to: " + uri);

    this._emitter = connect(uri);
    this._internalEmitter = new EventEmitter();

    this._emitter.on("connect", (...args) => {
      // Element is connected
      _this._logger.info("connected");
      _this.connected.setContent(true);
    });
    this._emitter.on("disconnect", () => {
      // Connection Lost.
      _this._logger.error("Connection lost!");
      _this.connected.setContent(false);
    });
  }

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

  async emitTaskCancelation(msg: ITaskCancelationMsg) {
    this._internalEmitter.emit("cancel", msg);
    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._internalEmitter.emit("generators", generators);
    this._emitter.emit("generators", generators);
  }

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

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

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

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

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

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

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

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

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

  async onBonjour(cb) {
    this._internalEmitter.on("bonjour", cb);
    this._emitter.on("bonjour", cb);
  }

  async emitBonjour(msg) {
    this._internalEmitter.emit("bonjour", msg);
    this._emitter.emit("bonjour", msg);
    this._logger.info("sending bonjour", msg);
  }

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

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

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

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

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

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

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