nope/lib/communication/IoSocketClient.ts

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

import { EventEmitter } from 'events';
import { connect, Socket } from 'socket.io-client';
import { Logger } from "winston";
import { getNopeLogger } from "../logger/getLogger";
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';

export class IoSocketClient implements ICommunicationInterface {

    connected: INopeObservable<boolean>;

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

    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: (dispatcher: string) => void) {
        this._internalEmitter.on('bonjour', cb);
        this._emitter.on('bonjour', cb);
    }

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

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

    async onNewTopicsAvailable(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);
    }
}