nope/lib/communication/IoSocketServer.ts

import { Server } from "http";
import * as io from 'socket.io';
import { Logger } from "winston";
import { getLogger } from "../logger/getLogger";
import { IAvailableInstanceGenerators, IAvailableServicesMsg, IAvailableTopicsMsg, ICommunicationInterface, IExternalEventMsg, IRequestTaskMsg, IResponseTaskMsg } from "../types/communication.interface";

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

    /**
     * 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._logger = getLogger('info', 'IO-Socket');

        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);
            });

            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._sockets = new Set();
        this._funcs = new Map<string, (...args: any[]) => void>();
    }
    off(event: string, cb: (...args: any[]) => 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) {
        // Add the Subscription to the Sockets.
        for (const socket of this._sockets) {
            socket.on(event, cb);
        }

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

    emit(event: string, data: any): void {
        if (this._logger.isDebugEnabled()){
            this._logger.debug('sending data on: ' + event);
        }        
        this._socket.emit(event, data);
    }
}


/**
 * Communication Layer using IO-Sockets.
 *
 * @export
 * @class IoSocketServer
 * @implements {ICommunicationInterface}
 */
export class IoSocketServer implements ICommunicationInterface {

    protected _emitter: IoSocketSeverEventEmitter

    constructor(
        public port: number,
        public readonly subscriptionMode: 'individual' | 'generic',
        public readonly resultSharing: 'individual' | 'generic',
        server?: Server,
        protected _logger?: Logger
      ){
        this._emitter = new IoSocketSeverEventEmitter(port, server);
      }
    
      async emitNewInstanceGeneratorsAvailable(generators: IAvailableInstanceGenerators) {
        this._emitter.emit('generators',generators);
      }
    
      async onNewInstanceGeneratorsAvailable(cb: (generators: IAvailableInstanceGenerators) => 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);
      }
    }