lib/client.js
function MQlobberClient(stream, options)
{
    EventEmitter.call(this);

    options = options || {};

    this._options = options;
    this.subs = new Map();
    this._matcher = new QlobberDedup(options);
    this.mux = new BPMux(stream, options);
    this._done = false;

    var ths = this;

    function error(err)
    {
        /*jshint validthis: true */
        ths.emit('error', err, this);
    }

    function warning(err)
    {
        /*jshint validthis: true */
        if (!ths.emit('warning', err, this))
        {
            console.error(err);
        }
    }
    this._warning = warning;

    this._unexpected_data = function ()
    {
        while (true)
        {
            if (this.read() === null)
            {
                break;
            }

            warning.call(this, new Error('unexpected data'));
        }
    };

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

    this.mux.on('end', function ()
    {
        ths.subs.clear();
    });

    var duplex = this.mux.multiplex(options);

    function end()
    {
        /*jshint validthis: true */
        error.call(this, new Error('ended before handshaken'));
    }

    duplex.on('end', end);
    duplex.on('error', error);
    duplex.on('readable', this._unexpected_data);

    this.mux.on('peer_multiplex', function (duplex)
    {
        duplex.on('error', error);
    });

    this.mux.on('handshake', function (dplex, hdata, delay)
    {
        if (dplex === duplex)
        {
            duplex.removeListener('end', end);
            ths.emit('handshake', hdata);
            return duplex.end(); // currently no data on initial duplex
        }

        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
        dplex.on('end', dend);
        dplex.on('error', dend);
        
        if (hdata.length < 1)
        {
            return warning.call(dplex, new Error('buffer too small'));
        }

        var info = {},
            flags = hdata.readUInt8(0, true),
            pos = 1,
            topic;

        info.single = !!(flags & 1);
        info.existing = !!(flags & 4);

        if (flags & 2)
        {
            pos += 8;

            if (hdata.length < pos)
            {
                return warning.call(dplex, new Error('buffer too small'));
            }

            info.expires = hdata.readUInt32BE(pos - 8, true) * thirty_two_bits +
                           hdata.readUInt32BE(pos - 4, true);
        }

        if (flags & 8)
        {
            pos += 8;

            if (hdata.length < pos)
            {
                return warning.call(dplex, new Error('buffer too small'));
            }

            info.size = hdata.readUInt32BE(pos - 8, true) * thirty_two_bits +
                        hdata.readUInt32BE(pos - 4, true);
        }

        info.topic = hdata.toString('utf8', pos);

        var called = false, handshake;
        if (info.single)
        {
            handshake = delay();
        }

        function done(err)
        {
            if (called)
            {
                return;
            }
            called = true;

            if (err)
            {
                warning.call(dplex, err);
            }

            if (handshake)
            {
                var hdata = new Buffer(1);
                hdata.writeUInt8(err ? 1 : 0);
                handshake(hdata);
            }

            dplex.end();
        }

        var handlers = ths._matcher.match(info.topic);

        if (handlers.size === 0)
        {
            return done(new Error('no handlers'));
        }

        for (var handler of handlers)
        {
            handler.call(ths, dplex, info, done);
            if (info.single)
            {
                break;
            }
        }
    });
}
function MQlobberClient(stream, options)
{
    EventEmitter.call(this);

    options = options || {};

    this._options = options;
    this.subs = new Map();
    this._matcher = new QlobberDedup(options);
    this.mux = new BPMux(stream, options);
    this._done = false;

    var ths = this;

    function error(err)
    {
        /*jshint validthis: true */
        ths.emit('error', err, this);
    }

    function warning(err)
    {
        /*jshint validthis: true */
        if (!ths.emit('warning', err, this))
        {
            console.error(err);
        }
    }
    this._warning = warning;

    this._unexpected_data = function ()
    {
        while (true)
        {
            if (this.read() === null)
            {
                break;
            }

            warning.call(this, new Error('unexpected data'));
        }
    };

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

    this.mux.on('end', function ()
    {
        ths.subs.clear();
    });

    var duplex = this.mux.multiplex(options);

    function end()
    {
        /*jshint validthis: true */
        error.call(this, new Error('ended before handshaken'));
    }

    duplex.on('end', end);
    duplex.on('error', error);
    duplex.on('readable', this._unexpected_data);

    this.mux.on('peer_multiplex', function (duplex)
    {
        duplex.on('error', error);
    });

    this.mux.on('handshake', function (dplex, hdata, delay)
    {
        if (dplex === duplex)
        {
            duplex.removeListener('end', end);
            ths.emit('handshake', hdata);
            return duplex.end(); // currently no data on initial duplex
        }

        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
        dplex.on('end', dend);
        dplex.on('error', dend);
        
        if (hdata.length < 1)
        {
            return warning.call(dplex, new Error('buffer too small'));
        }

        var info = {},
            flags = hdata.readUInt8(0, true),
            pos = 1,
            topic;

        info.single = !!(flags & 1);
        info.existing = !!(flags & 4);

        if (flags & 2)
        {
            pos += 8;

            if (hdata.length < pos)
            {
                return warning.call(dplex, new Error('buffer too small'));
            }

            info.expires = hdata.readUInt32BE(pos - 8, true) * thirty_two_bits +
                           hdata.readUInt32BE(pos - 4, true);
        }

        if (flags & 8)
        {
            pos += 8;

            if (hdata.length < pos)
            {
                return warning.call(dplex, new Error('buffer too small'));
            }

            info.size = hdata.readUInt32BE(pos - 8, true) * thirty_two_bits +
                        hdata.readUInt32BE(pos - 4, true);
        }

        info.topic = hdata.toString('utf8', pos);

        var called = false, handshake;
        if (info.single)
        {
            handshake = delay();
        }

        function done(err)
        {
            if (called)
            {
                return;
            }
            called = true;

            if (err)
            {
                warning.call(dplex, err);
            }

            if (handshake)
            {
                var hdata = new Buffer(1);
                hdata.writeUInt8(err ? 1 : 0);
                handshake(hdata);
            }

            dplex.end();
        }

        var handlers = ths._matcher.match(info.topic);

        if (handlers.size === 0)
        {
            return done(new Error('no handlers'));
        }

        for (var handler of handlers)
        {
            handler.call(ths, dplex, info, done);
            if (info.single)
            {
                break;
            }
        }
    });
}
function MQlobberClient ( stream , options ) { EventEmitter . call ( this ) ; options = options || { } ; this . _options = options ; this . subs = new Map ( ) ; this . _matcher = new QlobberDedup ( options ) ; this . mux = new BPMux ( stream , options ) ; this . _done = false ; var ths = this ; function error ( err ) { /*jshint validthis: true */ ths . emit ( 'error' , err , this ) ; } function warning ( err ) { /*jshint validthis: true */ if ( ! ths . emit ( 'warning' , err , this ) ) { console . error ( err ) ; } } this . _warning = warning ; this . _unexpected_data = function ( ) { while ( true ) { if ( this . read ( ) === null ) { break ; } warning . call ( this , new Error ( 'unexpected data' ) ) ; } } ; 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 ; } ) ; this . mux . on ( 'end' , function ( ) { ths . subs . clear ( ) ; } ) ; var duplex = this . mux . multiplex ( options ) ; function end ( ) { /*jshint validthis: true */ error . call ( this , new Error ( 'ended before handshaken' ) ) ; } duplex . on ( 'end' , end ) ; duplex . on ( 'error' , error ) ; duplex . on ( 'readable' , this . _unexpected_data ) ; this . mux . on ( 'peer_multiplex' , function ( duplex ) { duplex . on ( 'error' , error ) ; } ) ; this . mux . on ( 'handshake' , function ( dplex , hdata , delay ) { if ( dplex === duplex ) { duplex . removeListener ( 'end' , end ) ; ths . emit ( 'handshake' , hdata ) ; return duplex . end ( ) ; // currently no data on initial duplex } 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 dplex . on ( 'end' , dend ) ; dplex . on ( 'error' , dend ) ; if ( hdata . length < 1 ) { return warning . call ( dplex , new Error ( 'buffer too small' ) ) ; } var info = { } , flags = hdata . readUInt8 ( 0 , true ) , pos = 1 , topic ; info . single = ! ! ( flags & 1 ) ; info . existing = ! ! ( flags & 4 ) ; if ( flags & 2 ) { pos += 8 ; if ( hdata . length < pos ) { return warning . call ( dplex , new Error ( 'buffer too small' ) ) ; } info . expires = hdata . readUInt32BE ( pos - 8 , true ) * thirty_two_bits + hdata . readUInt32BE ( pos - 4 , true ) ; } if ( flags & 8 ) { pos += 8 ; if ( hdata . length < pos ) { return warning . call ( dplex , new Error ( 'buffer too small' ) ) ; } info . size = hdata . readUInt32BE ( pos - 8 , true ) * thirty_two_bits + hdata . readUInt32BE ( pos - 4 , true ) ; } info . topic = hdata . toString ( 'utf8' , pos ) ; var called = false , handshake ; if ( info . single ) { handshake = delay ( ) ; } function done ( err ) { if ( called ) { return ; } called = true ; if ( err ) { warning . call ( dplex , err ) ; } if ( handshake ) { var hdata = new Buffer ( 1 ) ; hdata . writeUInt8 ( err ? 1 : 0 ) ; handshake ( hdata ) ; } dplex . end ( ) ; } var handlers = ths . _matcher . match ( info . topic ) ; if ( handlers . size === 0 ) { return done ( new Error ( 'no handlers' ) ) ; } for ( var handler of handlers ) { handler . call ( ths , dplex , info , done ) ; if ( info . single ) { break ; } } } ) ; }
Create a new `MQlobberClient` object for publishing and subscribing to
messages via a server.

@constructor

@param {Duplex} stream Connection to a server. The server should use [`MQlobberServer`](#mqlobberserverfsq-stream-options) on its side of the connection. How the connection is made is up to the caller - it just has to supply a [`Duplex`](https://nodejs.org/dist/latest-v4.x/docs/api/stream.html#stream_class_stream_duplex). For example, [`net.Socket`](https://nodejs.org/dist/latest-v4.x/docs/api/net.html#net_class_net_socket) or [`PrimusDuplex`](https://github.com/davedoesdev/primus-backpressure#primusduplexmsg_stream-options).

@param {Object} [options] Configuration options. This is passed down to [`QlobberDedup`](https://github.com/davedoesdev/qlobber#qlobberdedupoptions) (which matches messages received from the server to handlers) and [`BPMux`](https://github.com/davedoesdev/bpmux#bpmuxcarrier-options) (which multiplexes message streams over the connection to the server). It also supports the following additional property:
- `{Buffer} [handshake_data]` Application-specific handshake data to send to the server. The server-side [`MQlobberServer`](#mqlobberserverfsq-stream-options) object will emit this as a [`handshake`](#mqlobberservereventshandshakehandshake_data-delay_handshake) event to its application.

@throws {Error} If an error occurs before initiating the multiplex with the server.
Create a new MQlobberClient object for publishing and subscribing to messages via a server .