function KafkaConsumerStream(consumer, options) {
  if (!(this instanceof KafkaConsumerStream)) {
    return new KafkaConsumerStream(consumer, options);
  }

  if (options === undefined) {
    options = { waitInterval: 1000 };
  } else if (typeof options === 'number') {
    options = { waitInterval: options };
  } else if (options === null || typeof options !== 'object') {
    throw new TypeError('"options" argument must be a number or an object');
  }

  var topics = options.topics;

  if (typeof topics === 'function') {
    // Just ignore the rest of the checks here
  } else if (!Array.isArray(topics)) {
    if (typeof topics !== 'string' && !(topics instanceof RegExp)) {
      throw new TypeError('"topics" argument must be a string, regex, or an array');
    } else {
      topics = [topics];
    }
  }

  options = Object.create(options);

  var fetchSize = options.fetchSize || 1;

  // Run in object mode by default.
  if (options.objectMode === null || options.objectMode === undefined) {
    options.objectMode = true;

    // If they did not explicitly set high water mark, and we are running
    // in object mode, set it to the fetch size + 2 to ensure there is room
    // for a standard fetch
    if (!options.highWaterMark) {
      options.highWaterMark = fetchSize + 2;
    }
  }

  if (options.objectMode !== true) {
    this._read = this._read_buffer;
  } else {
    this._read = this._read_message;
  }

  Readable.call(this, options);

  this.consumer = consumer;
  this.topics = topics;
  this.autoClose = options.autoClose === undefined ? true : !!options.autoClose;
  this.waitInterval = options.waitInterval === undefined ? 1000 : options.waitInterval;
  this.fetchSize = fetchSize;
  this.connectOptions = options.connectOptions || {};
  this.streamAsBatch = options.streamAsBatch || false;

  // Hold the messages in here
  this.messages = [];

  var self = this;

  this.consumer
    .on('unsubscribed', function() {
      // Invalidate the stream when we unsubscribe
      self.push(null);
    });

  if (options.initOauthBearerToken) {
    this.consumer.setOauthBearerToken(options.initOauthBearerToken);
  }

  // Call connect. Handles potentially being connected already
  this.connect(this.connectOptions);

  this.once('end', function() {
    if (this.autoClose) {
      this.destroy();
    }
  });

}
ReadableStream integrating with the Kafka Consumer.

This class is used to read data off of Kafka in a streaming way. It is
useful if you'd like to have a way to pipe Kafka into other systems. You
should generally not make this class yourself, as it is not even exposed
as part of module.exports. Instead, you should KafkaConsumer.createReadStream.

The stream implementation is slower than the continuous subscribe callback.
If you don't care so much about backpressure and would rather squeeze
out performance, use that method. Using the stream will ensure you read only
as fast as you write.

The stream detects if Kafka is already connected. If it is, it will begin
reading. If it is not, it will connect and read when it is ready.

This stream operates in objectMode. It streams {Consumer~Message}

@param {Consumer} consumer - The Kafka Consumer object.
@param {object} options - Options to configure the stream.
@param {number} options.waitInterval - Number of ms to wait if Kafka reports
that it has timed out or that we are out of messages (right now).
@param {array} options.topics - Array of topics, or a function that parses
metadata into an array of topics
@constructor
@extends stream.Readable
@see Consumer~Message