lib/server.js
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)
    {
        /*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`.