import { Base64Message } from './Model/Base64Message' ; import { DataSource , MqttSource } from './DataSource' ; import { AddMqttConnection , MqttMessage , addMqttConnectionEvent , backendEvents , makeConnectionMessageEvent , makeConnectionStateEvent , makePublishEvent , removeConnection , } from '../../events' ; export class ConnectionManager { private connections : { [ s ] : DataSource < any > ; } = { } ; private handleConnectionRequest = ( event ) => { const connectionId = event . id ; if ( this . connections [ connectionId ] ) { this . removeConnection ( connectionId ) ; } const options = event . options ; const connection = new MqttSource ( ) ; this . connections [ connectionId ] = connection ; const connectionStateEvent = makeConnectionStateEvent ( connectionId ) ; connection . stateMachine . onUpdate . subscribe ( state => { backendEvents . emit ( connectionStateEvent , state ) ; } ) ; connection . connect ( options ) ; this . handleNewMessagesForConnection ( connectionId , connection ) ; backendEvents . subscribe ( makePublishEvent ( connectionId ) , ( msg ) => { this . connections [ connectionId ] . publish ( msg ) ; } ) ; } ; private handleNewMessagesForConnection ( connectionId , connection ) { const messageEvent = makeConnectionMessageEvent ( connectionId ) ; connection . onMessage ( ( topic , payload , packet ) => { let buffer = payload ; if ( buffer . length > 20000 ) { buffer = buffer . slice ( 0 , 20000 ) ; } backendEvents . emit ( messageEvent , { topic , payload : Base64Message . fromBuffer ( buffer ) , qos : packet . qos , retain : packet . retain , messageId : packet . messageId } ) ; } ) ; } public manageConnections ( ) { backendEvents . subscribe ( addMqttConnectionEvent , this . handleConnectionRequest ) ; backendEvents . subscribe ( removeConnection , ( connectionId ) => { this . removeConnection ( connectionId ) ; } ) ; } public removeConnection ( connectionId ) { const connection = this . connections [ connectionId ] ; if ( connection ) { backendEvents . unsubscribeAll ( makePublishEvent ( connectionId ) ) ; connection . disconnect ( ) ; delete this . connections [ connectionId ] ; connection . stateMachine . onUpdate . removeAllListeners ( ) ; } } public closeAllConnections ( ) { Object . keys ( this . connections ) . forEach ( connectionId => this . removeConnection ( connectionId ) ) ; } }