import { EventEmitter } from 'events' ; import { log } from '../utils/logger' ; export function Client ( params , emitter , handle ? ) { let connected = 1 ; const channels = { } ; const routine = params . routine ( params , new EventEmitter ( ) ) ; const socket = params . transport ( params , emitter ) ; let totalMessages = 0 ; let instance ; const remote = ( params . isServer ) ? socket . remote ( handle ) : { host : params . host , port : params . port } ; const local = ( params . isServer ) ? { host : params . host , port : params . port } : null ; function _resolveChannel ( channelName ) { if ( ! ( channelName in channels ) ) { channels [ channelName ] = { name : channelName , packets : [ ] , handlers : [ ] , } ; } return channels [ channelName ] ; } function getChannels ( ) { return Object . keys ( channels ) ; } function _wrap ( event ) { socket . send ( handle , getChannels ( ) . reduce ( ( frame , channelName ) => { frame . channels [ channelName ] = channels [ channelName ] . packets ; return frame ; } , { frameId : event . frameId , channels : { } } ) ) ; getChannels ( ) . forEach ( channelName => { channels [ channelName ] . packets . length = 0 ; } ) ; totalMessages = 0 ; } function _handleConnect ( ) { connected = 2 ; log ( `connected to ${ params . host }:${ params . port }` ) ; } function _handleError ( err ) { log ( `error ${ err . message }` ) ; } function _handleRequest ( frame , payloadBytes ) { if ( frame && frame . channels ) { Object . keys ( frame . channels ) . forEach ( channelName => { frame . channels [ channelName ] . forEach ( ( packet , messageIndex ) => { if ( channelName in channels ) { channels [ channelName ] . handlers . forEach ( handler => handler ( packet , { client : instance , frame : { channel : channelName , id : frame . frameId , messageIndex , payloadBytes , payloadMessages : frame . channels [ channelName ] . length , } , } ) ) ; } } ) ; } ) ; } } function _handleDisconnect ( ) { connected = 0 ; log ( `lost connection to ${ params . host }:${ params . port }` ) ; } function write ( channelName , message ) { if ( params . json !== true && ! Buffer . isBuffer ( message ) ) { throw new Error ( `Unable to serialize message: ${ message }, expected type Buffer` ) ; } _resolveChannel ( channelName ) . packets . push ( message ) ; if ( ++ totalMessages >= 0xff ) { log ( 'Buffer saturated, flushing messages to transport' ) ; routine . flush ( ) ; } else routine . add ( message ) ; } function destroy ( ) { routine . flush ( ) ; if ( connected > 1 ) setTimeout ( ( ) => socket . disconnect ( handle ) , 0 ) ; } function subscribe ( channel , handler ) { _resolveChannel ( channel ) . handlers . push ( handler ) ; } function unsubscribe ( channelName , handler ? ) { if ( ! ( channelName in channels ) ) return ; if ( handler ) { const index = channels [ channelName ] . handlers . indexOf ( handler ) ; if ( index > - 1 ) channels [ channelName ] . handlers . splice ( index , 1 ) ; } else channels [ channelName ] . handlers = [ ] ; if ( channels [ channelName ] . handlers . length === 0 && channels [ channelName ] . packets . length === 0 ) delete channels [ channelName ] ; } routine . emitter . on ( 'runQueue' , _wrap ) ; emitter . on ( 'connect' , _handleConnect ) ; emitter . on ( 'disconnect' , _handleDisconnect ) ; emitter . on ( 'error' , _handleError ) ; emitter . on ( 'frame' , _handleRequest ) ; if ( ! handle ) log ( `connecting to ${ params . host }:${ params . port }` ) ; handle = socket . connect ( handle ) ; instance = Object . assign ( emitter , { write , destroy , subscribe , unsubscribe , remote , local , label : params . label , getChannels , } ) ; return instance ; }