/** * @author Martin Karkowski * @email m.karkowski@zema.de * @create date 2020-11-06 08:52:42 * @modify date 2020-11-06 08:52:43 * @desc [description] */ import { Server } from "http"; import * as io from 'socket.io'; import { getNopeLogger } from "../logger/getLogger"; import { Logger } from "winston"; import { IAvailableInstanceGeneratorsMsg, IAvailableServicesMsg, IAvailableTopicsMsg, ICommunicationInterface, IExternalEventMsg, IRequestTaskMsg, IResponseTaskMsg, ITaskCancelationMsg } from "../types/nope/nopeCommunication.interface"; /** * Wrapper Interface. * Creates an Interface for a Socket server */ export class IoSocketSeverEventEmitter { protected _socket: io.Server protected _sockets: Set; protected _funcs: Map 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 = getNopeLogger('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 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 onTaskCancelation(cb: (msg: ITaskCancelationMsg) => void) { this._emitter.on('cancel',cb); } async emitTaskCancelation(msg: ITaskCancelationMsg) { 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._emitter.emit('generators',generators); } async onNewInstanceGeneratorsAvailable(cb: (generators: IAvailableInstanceGeneratorsMsg) => 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); } }