lib/producer-stream.js
function ProducerStream(producer, options) {
  if (!(this instanceof ProducerStream)) {
    return new ProducerStream(producer, options);
  }

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

  if (!options.objectMode && !options.topic) {
    throw new TypeError('ProducerStreams not using objectMode must provide a topic to produce to.');
  }

  if (options.objectMode !== true) {
    this._write = this._write_buffer;
  } else {
    this._write = this._write_message;
  }

  Writable.call(this, options);

  this.producer = producer;
  this.topicName = options.topic;

  this.autoClose = options.autoClose === undefined ? true : !!options.autoClose;
  this.connectOptions = options.connectOptions || {};

  this.producer.setPollInterval(options.pollInterval || 1000);

  if (options.encoding) {
    this.setDefaultEncoding(options.encoding);
  }

  // Connect to the producer. Unless we are already connected
  if (!this.producer.isConnected()) {
    this.connect(this.connectOptions);
  }

  var self = this;

  this.once('finish', function() {
    if (this.autoClose) {
      this.close();
    }
  });

}
function ProducerStream(producer, options) {
  if (!(this instanceof ProducerStream)) {
    return new ProducerStream(producer, options);
  }

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

  if (!options.objectMode && !options.topic) {
    throw new TypeError('ProducerStreams not using objectMode must provide a topic to produce to.');
  }

  if (options.objectMode !== true) {
    this._write = this._write_buffer;
  } else {
    this._write = this._write_message;
  }

  Writable.call(this, options);

  this.producer = producer;
  this.topicName = options.topic;

  this.autoClose = options.autoClose === undefined ? true : !!options.autoClose;
  this.connectOptions = options.connectOptions || {};

  this.producer.setPollInterval(options.pollInterval || 1000);

  if (options.encoding) {
    this.setDefaultEncoding(options.encoding);
  }

  // Connect to the producer. Unless we are already connected
  if (!this.producer.isConnected()) {
    this.connect(this.connectOptions);
  }

  var self = this;

  this.once('finish', function() {
    if (this.autoClose) {
      this.close();
    }
  });

}
function ProducerStream ( producer , options ) { if ( ! ( this instanceof ProducerStream ) ) { return new ProducerStream ( producer , options ) ; } if ( options === undefined ) { options = { } ; } else if ( typeof options === 'string' ) { options = { encoding : options } ; } else if ( options === null || typeof options !== 'object' ) { throw new TypeError ( '"streamOptions" argument must be a string or an object' ) ; } if ( ! options . objectMode && ! options . topic ) { throw new TypeError ( 'ProducerStreams not using objectMode must provide a topic to produce to.' ) ; } if ( options . objectMode !== true ) { this . _write = this . _write_buffer ; } else { this . _write = this . _write_message ; } Writable . call ( this , options ) ; this . producer = producer ; this . topicName = options . topic ; this . autoClose = options . autoClose === undefined ? true : ! ! options . autoClose ; this . connectOptions = options . connectOptions || { } ; this . producer . setPollInterval ( options . pollInterval || 1000 ) ; if ( options . encoding ) { this . setDefaultEncoding ( options . encoding ) ; } // Connect to the producer. Unless we are already connected if ( ! this . producer . isConnected ( ) ) { this . connect ( this . connectOptions ) ; } var self = this ; this . once ( 'finish' , function ( ) { if ( this . autoClose ) { this . close ( ) ; } } ) ; }
Writable stream integrating with the Kafka Producer.

This class is used to write data to Kafka in a streaming way. It takes
buffers of data and puts them into the appropriate Kafka topic. If you need
finer control over partitions or keys, this is probably not the class for
you. In that situation just use the Producer itself.

The stream detects if Kafka is already connected. You can safely begin
writing right away.

This stream does not operate in Object mode and can only be given buffers.

@param {Producer} producer - The Kafka Producer object.
@param {array} topics - Array of topics
@param {object} options - Topic configuration.
@constructor
@extends stream.Writable
Writable stream integrating with the Kafka Producer .