whole_func_string: 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;
              }
          }
      });
  }
func_code_string: 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;
              }
          }
      });
  }
func_documentation_string: 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.