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