nl_tokens: Create a new MQlobberServer object for publishing and subscribing to messages on behalf of a client .
pl_tokens: function MQlobberServer ( fsq , stream , options ) { EventEmitter . call ( this ) ; options = options || { } ; this . fsq = fsq ; this . subs = new Set ( ) ; this . _done = <mask> ; this . mux = new BPMux ( stream , util . _extend ( util . _extend ( { } , options ) , { high_channels : true } ) ) ; var ths = this ; function error ( err ) { /*jshint validthis: true */ ths . emit ( 'error' , err , this ) ; } function warning ( err ) { /*jshint validthis: true */ if ( err && ! ths . emit ( 'warning' , err , this ) ) { console . error ( err ) ; } } this . _warning = warning ; this . mux . on ( 'error' , error ) ; function handshake_sent ( duplex , complete ) { if ( ! complete ) { ths . emit ( 'backoff' ) ; } } this . mux . on ( 'handshake_sent' , handshake_sent ) ; this . mux . on ( 'pre_handshake_sent' , handshake_sent ) ; this . mux . on ( 'drain' , function ( ) { ths . emit ( 'drain' ) ; } ) ; this . mux . on ( 'full' , function ( ) { ths . emit ( 'full' ) ; } ) ; this . mux . on ( 'removed' , function ( duplex ) { ths . emit ( 'removed' , duplex ) ; } ) ; this . mux . on ( 'finish' , function ( ) { ths . _done = true ; if ( ! ths . emit ( 'unsubscribe_all_requested' , warning ) ) { ths . default_unsubscribe_all_requested_handler ( ) ; } } ) ; this . relay_error = function ( err ) { /*jshint validthis: true */ switch ( this . _readableState . pipesCount ) { case 0 : break ; case 1 : this . _readableState . pipes . emit ( 'error' , err ) ; break ; default : var dests = new Set ( ) , dest ; for ( dest of this . _readableState . pipes ) { dests . add ( dest ) ; } for ( dest of dests ) { dest . emit ( 'error' , err ) ; } break ; } } ; this . handler = function ( data , info , cb3 ) { var called = false ; function cb ( err , cb2 ) { warning ( err ) ; if ( ! called ) { cb3 . num_handlers -= 1 ; called = true ; } if ( options . defer_to_final_handler && ( cb3 . num_handlers > 0 ) ) { if ( cb2 ) { return cb2 ( err ) ; } return ; } cb3 ( err , cb2 ) ; } data . on ( 'error' , warning ) ; data . on ( 'error' , ths . relay_error ) ; if ( ! info . single ) { data . on ( 'end' , cb ) ; } var hdata = [ new Buffer ( 1 ) ] , buf ; hdata [ 0 ] . writeUInt8 ( ( info . single ? 1 : 0 ) | ( ( options . send_expires ? 1 : 0 ) << 1 ) | ( ( info . existing ? 1 : 0 ) << 2 ) | ( ( options . send_size ? 1 : 0 ) << 3 ) , 0 ) ; if ( options . send_expires ) { buf = new Buffer ( 8 ) ; var expires = Math . floor ( info . expires / 1000 ) ; // 32 bits are only enough for dates up to 2106 // can't use >> and & because JS converts to 32 bits for bitwise ops buf . writeUInt32BE ( Math . floor ( expires / thirty_two_bits ) , 0 ) ; buf . writeUInt32BE ( expires % thirty_two_bits , 4 ) ; hdata . push ( buf ) ; } if ( options . send_size ) { buf = new Buffer ( 8 ) ; buf . writeUInt32BE ( Math . floor ( info . size / thirty_two_bits ) , 0 ) ; buf . writeUInt32BE ( info . size % thirty_two_bits , 4 ) ; hdata . push ( buf ) ; } hdata . push ( new Buffer ( info . topic , 'utf8' ) ) ; function multiplex ( on_error ) { var duplex = ths . mux . multiplex ( Object . assign ( { } , options , { handshake_data : Buffer . concat ( hdata ) } ) ) ; duplex . on ( 'error' , function ( ) { this . peer_error_then_end ( ) ; } ) ; duplex . on ( 'error' , on_error || cb ) ; duplex . on ( 'readable' , function ( ) { while ( true ) { if ( this . read ( ) === null ) { break ; } warning . call ( this , new Error ( 'unexpected data' ) ) ; } } ) ; if ( info . single ) { var end = function ( ) { cb ( new Error ( 'ended before handshaken' ) ) ; } ; duplex . on ( 'end' , end ) ; duplex . on ( 'handshake' , function ( hdata ) { this . removeListener ( 'end' , end ) ; if ( hdata . length < 1 ) { return cb ( new Error ( 'buffer too small' ) ) ; } if ( hdata . readUInt8 ( 0 , true ) !== 0 ) { return cb ( new Error ( 'client error' ) ) ; } duplex . emit ( 'ack' , info ) ; ths . emit ( 'ack' , info ) ; cb ( ) ; } ) ; } return duplex ; } multiplex . num_handlers = cb3 . num_handlers ; if ( ! ths . emit ( 'message' , data , info , multiplex , cb ) ) { ths . default_message_handler ( data , info , multiplex , cb ) ; } } ; this . handler . accept_stream = true ; this . handler . mqlobber_server = this ; function end ( ) { /*jshint validthis: true */ error . call ( this , new Error ( 'ended before handshaken' ) ) ; } this . mux . on ( 'end' , end ) ; this . mux . on ( 'peer_multiplex' , function ( duplex ) { duplex . on ( 'error' , warning ) ; } ) ; this . mux . once ( 'handshake' , function ( duplex , hdata , delay ) { this . removeListener ( 'end' , end ) ; duplex . end ( ) ; // currently no data on initial duplex this . on ( 'handshake' , function ( duplex , hdata , delay ) { if ( ! delay ) { // duplex was initiated by us (outgoing message) return ; } function dend ( ) { /*jshint validthis: true */ this . end ( ) ; } // only read from incoming duplex but don't end until all data is // read in order to allow application to apply back-pressure duplex . on ( 'end' , dend ) ; duplex . on ( 'error' , dend ) ; duplex . on ( 'error' , ths . relay_error ) ; var handshake = delay ( ) ; function done ( err ) { warning . call ( duplex , err ) ; var hdata = new Buffer ( 1 ) ; hdata . writeUInt8 ( err ? 1 : 0 ) ; if ( arguments . length > 1 ) { var data = arguments [ arguments . length - 1 ] ; if ( Buffer . isBuffer ( data ) ) { hdata = Buffer . concat ( [ hdata , data ] ) ; } } handshake ( hdata ) ; duplex . on ( 'readable' , function ( ) { while ( true ) { if ( this . read ( ) === null ) { break ; } warning . call ( this , new Error ( 'unexpected data' ) ) ; } } ) ; } if ( hdata . length < 1 ) { return done ( new Error ( 'buffer too small' ) ) ; } var type = hdata . readUInt8 ( 0 , true ) , topic ; switch ( type ) { case TYPE_SUBSCRIBE : topic = hdata . toString ( 'utf8' , 1 ) ; if ( ! ths . emit ( 'pre_subscribe_requested' , topic , done ) && ! ths . emit ( 'subscribe_requested' , topic , done ) ) { ths . default_subscribe_requested_handler ( topic , done ) ; } break ; case TYPE_UNSUBSCRIBE : topic = hdata . toString ( 'utf8' , 1 ) ; if ( ! ths . emit ( 'pre_unsubscribe_requested' , topic , done ) && ! ths . emit ( 'unsubscribe_requested' , topic , done ) ) { ths . default_unsubscribe_requested_handler ( topic , done ) ; } break ; case TYPE_UNSUBSCRIBE_ALL : if ( ! ths . emit ( 'unsubscribe_all_requested' , done ) ) { ths . default_unsubscribe_all_requested_handler ( done ) ; } break ; case TYPE_PUBLISH : if ( hdata . length < 2 ) { return done ( new Error ( 'buffer too small' ) ) ; } var options = { } , flags = hdata . readUInt8 ( 1 , true ) , pos = 2 ; options . single = ! ! ( flags & 1 ) ; if ( flags & 2 ) { pos += 4 ; if ( hdata . length < pos ) { return done ( new Error ( 'buffer too small' ) ) ; } options . ttl = Math . min ( options . single ? fsq . _single_ttl : fsq . _multi_ttl , hdata . readUInt32BE ( 2 , true ) * 1000 ) ; } topic = hdata . toString ( 'utf8' , pos ) ; if ( ! ths . emit ( 'pre_publish_requested' , topic , duplex , options , done ) && ! ths . emit ( 'publish_requested' , topic , duplex , options , done ) ) { ths . default_publish_requested_handler ( topic , duplex , options , done ) ; } break ; default : done ( new Error ( 'unknown type: ' + type ) ) ; break ; } } ) ; ths . emit ( 'handshake' , hdata , delay ) ; } ) ; }