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

  // Force this to be true for the high level producer
  conf = shallowCopy(conf);
  conf.dr_cb = true;

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

  // Add a delivery emitter to the producer
  this._hl = {
    deliveryEmitter: new EventEmitter(),
    messageId: 0,
    // Special logic for polling. We use a reference counter to know when we need
    // to be doing it and when we can stop. This means when we go into fast polling
    // mode we don't need to do multiple calls to poll since they all will yield
    // the same result
    pollingRefTimeout: null,
  };

  // Add the polling ref counter to the class which ensures we poll when we go active
  this._hl.pollingRef = new RefCounter(function() {
    self._hl.pollingRefTimeout = setInterval(function() {
      try {
        self.poll();
      } catch (e) {
        if (!self._isConnected) {
          // If we got disconnected for some reason there is no point
          // in polling anymore
          clearInterval(self._hl.pollingRefTimeout);
        }
      }
    }, 1);
  }, function() {
    clearInterval(self._hl.pollingRefTimeout);
  });

  // Default poll interval. More sophisticated polling is also done in create rule method
  this.setPollInterval(1000);

  // Listen to all delivery reports to propagate elements with a _message_id to the emitter
  this.on('delivery-report', function(err, report) {
    if (report.opaque && report.opaque.__message_id !== undefined) {
      self._hl.deliveryEmitter.emit(report.opaque.__message_id, err, report.offset);
    }
  });

  // Save old produce here since we are making some modifications for it
  this._oldProduce = this.produce;
  this.produce = this._modifiedProduce;

  // Serializer information
  this.keySerializer = noopSerializer;
  this.valueSerializer = noopSerializer;
}
function HighLevelProducer(conf, topicConf) {
  if (!(this instanceof HighLevelProducer)) {
    return new HighLevelProducer(conf, topicConf);
  }

  // Force this to be true for the high level producer
  conf = shallowCopy(conf);
  conf.dr_cb = true;

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

  // Add a delivery emitter to the producer
  this._hl = {
    deliveryEmitter: new EventEmitter(),
    messageId: 0,
    // Special logic for polling. We use a reference counter to know when we need
    // to be doing it and when we can stop. This means when we go into fast polling
    // mode we don't need to do multiple calls to poll since they all will yield
    // the same result
    pollingRefTimeout: null,
  };

  // Add the polling ref counter to the class which ensures we poll when we go active
  this._hl.pollingRef = new RefCounter(function() {
    self._hl.pollingRefTimeout = setInterval(function() {
      try {
        self.poll();
      } catch (e) {
        if (!self._isConnected) {
          // If we got disconnected for some reason there is no point
          // in polling anymore
          clearInterval(self._hl.pollingRefTimeout);
        }
      }
    }, 1);
  }, function() {
    clearInterval(self._hl.pollingRefTimeout);
  });

  // Default poll interval. More sophisticated polling is also done in create rule method
  this.setPollInterval(1000);

  // Listen to all delivery reports to propagate elements with a _message_id to the emitter
  this.on('delivery-report', function(err, report) {
    if (report.opaque && report.opaque.__message_id !== undefined) {
      self._hl.deliveryEmitter.emit(report.opaque.__message_id, err, report.offset);
    }
  });

  // Save old produce here since we are making some modifications for it
  this._oldProduce = this.produce;
  this.produce = this._modifiedProduce;

  // Serializer information
  this.keySerializer = noopSerializer;
  this.valueSerializer = noopSerializer;
}
function HighLevelProducer ( conf , topicConf ) { if ( ! ( this instanceof HighLevelProducer ) ) { return new HighLevelProducer ( conf , topicConf ) ; } // Force this to be true for the high level producer conf = shallowCopy ( conf ) ; conf . dr_cb = true ; // producer is an initialized consumer object // @see NodeKafka::Producer::Init Producer . call ( this , conf , topicConf ) ; var self = this ; // Add a delivery emitter to the producer this . _hl = { deliveryEmitter : new EventEmitter ( ) , messageId : 0 , // Special logic for polling. We use a reference counter to know when we need // to be doing it and when we can stop. This means when we go into fast polling // mode we don't need to do multiple calls to poll since they all will yield // the same result pollingRefTimeout : null , } ; // Add the polling ref counter to the class which ensures we poll when we go active this . _hl . pollingRef = new RefCounter ( function ( ) { self . _hl . pollingRefTimeout = setInterval ( function ( ) { try { self . poll ( ) ; } catch ( e ) { if ( ! self . _isConnected ) { // If we got disconnected for some reason there is no point // in polling anymore clearInterval ( self . _hl . pollingRefTimeout ) ; } } } , 1 ) ; } , function ( ) { clearInterval ( self . _hl . pollingRefTimeout ) ; } ) ; // Default poll interval. More sophisticated polling is also done in create rule method this . setPollInterval ( 1000 ) ; // Listen to all delivery reports to propagate elements with a _message_id to the emitter this . on ( 'delivery-report' , function ( err , report ) { if ( report . opaque && report . opaque . __message_id !== undefined ) { self . _hl . deliveryEmitter . emit ( report . opaque . __message_id , err , report . offset ) ; } } ) ; // Save old produce here since we are making some modifications for it this . _oldProduce = this . produce ; this . produce = this . _modifiedProduce ; // Serializer information this . keySerializer = noopSerializer ; this . valueSerializer = noopSerializer ; }
Producer class for sending messages to Kafka in a higher level fashion

This is the main entry point for writing data to Kafka if you want more
functionality than librdkafka supports out of the box. 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.

This has a few restrictions, so it is not for free!

1. You may not define opaque tokens
The higher level producer is powered by opaque tokens.
2. Every message ack will dispatch an event on the node thread.
3. Will use a ref counter to determine if there are outgoing produces.

This will return the new object you should use instead when doing your
produce calls

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