function MQlobberServer(fsq, stream, options)
{
    EventEmitter.call(this);

    options = options || {};

    this.fsq = fsq;
    this.subs = new Set();
    this._done = false;
    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);
    });
}
function MQlobberServer(fsq, stream, options)
{
    EventEmitter.call(this);

    options = options || {};

    this.fsq = fsq;
    this.subs = new Set();
    this._done = false;
    this.mux = new BPMux(stream, util._extend(util._extend({}, options),
               {
                   high_channels: true
               }));

    var ths = this;
        
    function error(err)
    {
        
        ths.emit('error', err, this);
    }

    function warning(err)
    {
        
        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)
    {
        

        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);
            
            
            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()
    {
        
        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(); 

        this.on('handshake', function (duplex, hdata, delay)
        {
            if (!delay)
            {
                
                return;
            }

            function dend()
            {
                
                this.end();
            }

            
            
            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);
    });
}
function MQlobberServer ( fsq , stream , options ) { EventEmitter . call ( this ) ; options = options || { } ; this . fsq = fsq ; this . subs = new Set ( ) ; this . _done = false ; 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 ) ; } ) ; }
Create a new `MQlobberServer` object for publishing and subscribing to messages
on behalf of a client.

@constructor

@param {QlobberFSQ} fsq File system queue - an instance of [`QlobberFSQ`](https://github.com/davedoesdev/qlobber-fsq#qlobberfsqoptions).  This does the heavy-lifting of reading and writing messages to a directory on the file system.

@param {Duplex} stream Connection to the client. The client should use [`MQlobberClient`](#mqlobberclientstream-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 [`BPMux`](https://github.com/davedoesdev/bpmux#bpmuxcarrier-options) (which multiplexes message streams over the connection to the client). It also supports the following additional properties:
- `{Boolean} send_expires` Whether to include message expiry time in metadata sent to the client. Defaults to `false`.

- `{Boolean} send_size` Whether to include message size in metadata sent to then client. Defaults to `false`.

- `{Boolean} defer_to_final_handler` If `true` then a message stream is only considered finished when all `MQlobberServer` objects finish processing it. Defaults to `false`.
Create a new MQlobberServer object for publishing and subscribing to messages on behalf of a client .