lib/producer.js
function Producer(conf, topicConf) {
  if (!(this instanceof Producer)) {
    return new Producer(conf, topicConf);
  }

  conf = shallowCopy(conf);
  topicConf = shallowCopy(topicConf);

  /**
   * Producer message. This is sent to the wrapper, not received from it
   *
   * @typedef {object} Producer~Message
   * @property {string|buffer} message - The buffer to send to Kafka.
   * @property {Topic} topic - The Kafka topic to produce to.
   * @property {number} partition - The partition to produce to. Defaults to
   * the partitioner
   * @property {string} key - The key string to use for the message.
   * @see Consumer~Message
   */

  var gTopic = conf.topic || false;
  var gPart = conf.partition || null;
  var dr_cb = conf.dr_cb || null;
  var dr_msg_cb = conf.dr_msg_cb || null;

  // delete keys we don't want to pass on
  delete conf.topic;
  delete conf.partition;

  delete conf.dr_cb;
  delete conf.dr_msg_cb;

  // client is an initialized consumer object
  // @see NodeKafka::Producer::Init
  Client.call(this, conf, Kafka.Producer, topicConf);
  var self = this;

  // Delete these keys after saving them in vars
  this.globalConfig = conf;
  this.topicConfig = topicConf;
  this.defaultTopic = gTopic || null;
  this.defaultPartition = gPart == null ? -1 : gPart;

  this.sentMessages = 0;

  this.pollInterval = undefined;

  if (dr_msg_cb || dr_cb) {
    this._client.onDeliveryReport(function onDeliveryReport(err, report) {
      if (err) {
        err = LibrdKafkaError.create(err);
      }
      self.emit('delivery-report', err, report);
    }, !!dr_msg_cb);

    if (typeof dr_cb === 'function') {
      self.on('delivery-report', dr_cb);
    }

  }
}
function Producer(conf, topicConf) {
  if (!(this instanceof Producer)) {
    return new Producer(conf, topicConf);
  }

  conf = shallowCopy(conf);
  topicConf = shallowCopy(topicConf);

  /**
   * Producer message. This is sent to the wrapper, not received from it
   *
   * @typedef {object} Producer~Message
   * @property {string|buffer} message - The buffer to send to Kafka.
   * @property {Topic} topic - The Kafka topic to produce to.
   * @property {number} partition - The partition to produce to. Defaults to
   * the partitioner
   * @property {string} key - The key string to use for the message.
   * @see Consumer~Message
   */

  var gTopic = conf.topic || false;
  var gPart = conf.partition || null;
  var dr_cb = conf.dr_cb || null;
  var dr_msg_cb = conf.dr_msg_cb || null;

  // delete keys we don't want to pass on
  delete conf.topic;
  delete conf.partition;

  delete conf.dr_cb;
  delete conf.dr_msg_cb;

  // client is an initialized consumer object
  // @see NodeKafka::Producer::Init
  Client.call(this, conf, Kafka.Producer, topicConf);
  var self = this;

  // Delete these keys after saving them in vars
  this.globalConfig = conf;
  this.topicConfig = topicConf;
  this.defaultTopic = gTopic || null;
  this.defaultPartition = gPart == null ? -1 : gPart;

  this.sentMessages = 0;

  this.pollInterval = undefined;

  if (dr_msg_cb || dr_cb) {
    this._client.onDeliveryReport(function onDeliveryReport(err, report) {
      if (err) {
        err = LibrdKafkaError.create(err);
      }
      self.emit('delivery-report', err, report);
    }, !!dr_msg_cb);

    if (typeof dr_cb === 'function') {
      self.on('delivery-report', dr_cb);
    }

  }
}
function Producer ( conf , topicConf ) { if ( ! ( this instanceof Producer ) ) { return new Producer ( conf , topicConf ) ; } conf = shallowCopy ( conf ) ; topicConf = shallowCopy ( topicConf ) ; /**
   * Producer message. This is sent to the wrapper, not received from it
   *
   * @typedef {object} Producer~Message
   * @property {string|buffer} message - The buffer to send to Kafka.
   * @property {Topic} topic - The Kafka topic to produce to.
   * @property {number} partition - The partition to produce to. Defaults to
   * the partitioner
   * @property {string} key - The key string to use for the message.
   * @see Consumer~Message
   */ var gTopic = conf . topic || false ; var gPart = conf . partition || null ; var dr_cb = conf . dr_cb || null ; var dr_msg_cb = conf . dr_msg_cb || null ; // delete keys we don't want to pass on delete conf . topic ; delete conf . partition ; delete conf . dr_cb ; delete conf . dr_msg_cb ; // client is an initialized consumer object // @see NodeKafka::Producer::Init Client . call ( this , conf , Kafka . Producer , topicConf ) ; var self = this ; // Delete these keys after saving them in vars this . globalConfig = conf ; this . topicConfig = topicConf ; this . defaultTopic = gTopic || null ; this . defaultPartition = gPart == null ? - 1 : gPart ; this . sentMessages = 0 ; this . pollInterval = undefined ; if ( dr_msg_cb || dr_cb ) { this . _client . onDeliveryReport ( function onDeliveryReport ( err , report ) { if ( err ) { err = LibrdKafkaError . create ( err ) ; } self . emit ( 'delivery-report' , err , report ) ; } , ! ! dr_msg_cb ) ; if ( typeof dr_cb === 'function' ) { self . on ( 'delivery-report' , dr_cb ) ; } } }
Producer class for sending messages to Kafka

This is the main entry point for writing data to Kafka. You
configure this like you do any other client, with a global
configuration and default topic configuration.

Once you instantiate this object, you need to connect to it first.
This allows you to get the metadata and make sure the connection
can be made before you depend on it. After that, problems with
the connection will by brought down by using poll, which automatically
runs when a transaction is made on the object.

@param {object} conf - Key value pairs to configure the producer
@param {object} topicConf - Key value pairs to create a default
topic configuration
@extends Client
@constructor
Producer class for sending messages to Kafka