lib/client.js
function Client(globalConf, SubClientType, topicConf) {
  if (!(this instanceof Client)) {
    return new Client(globalConf, SubClientType, topicConf);
  }

  Emitter.call(this);

  // This superclass must be initialized with the Kafka.{Producer,Consumer}
  // @example var client = new Client({}, Kafka.Producer);
  // remember this is a superclass so this will get taken care of in
  // the producer and consumer main wrappers

  var no_event_cb = globalConf.event_cb === false;
  topicConf = topicConf || {};

  // delete this because librdkafka will complain since this particular
  // key is a real conf value
  delete globalConf.event_cb;

  this._client = new SubClientType(globalConf, topicConf);

  // primitive callbacks from c++ wrapper
  // need binds on them if we want to broadcast the event via the emitter

  // However, the C++ wrapper lazy processes callbacks.
  // If we want to improve performance, removing this line is a good way to do it.
  // we need to do this on creation of the object
  // thats why we did some processing on config

  // Self required because we are inside an async callback function scope
  if (!no_event_cb) {
    this._client.onEvent(function eventHandler(eventType, eventData) {
      switch (eventType) {
        case 'error':
          self.emit('event.error', LibrdKafkaError.create(eventData));
          break;
        case 'stats':
          self.emit('event.stats', eventData);
          break;
        case 'log':
          self.emit('event.log', eventData);
          break;
        default:
          self.emit('event.event', eventData);
          self.emit('event.' + eventType, eventData);
      }
    });
  }

  this.metrics = {};
  this._isConnected = false;
  this.errorCounter = 0;

  /**
   * Metadata object. Starts out empty but will be filled with information after
   * the initial connect.
   *
   * @type {Client~Metadata}
   */
  this._metadata = {};

  var self = this;

  this.on('ready', function(info) {
    self.metrics.connectionOpened = Date.now();
    self.name = info.name;
  })
  .on('disconnected', function() {
    // reset metrics
    self.metrics = {};
    self._isConnected = false;
    // keep the metadata. it still may be useful
  })
  .on('event.error', function(err) {
    self.lastError = err;
    ++self.errorCounter;
  });

}
function Client(globalConf, SubClientType, topicConf) {
  if (!(this instanceof Client)) {
    return new Client(globalConf, SubClientType, topicConf);
  }

  Emitter.call(this);

  // This superclass must be initialized with the Kafka.{Producer,Consumer}
  // @example var client = new Client({}, Kafka.Producer);
  // remember this is a superclass so this will get taken care of in
  // the producer and consumer main wrappers

  var no_event_cb = globalConf.event_cb === false;
  topicConf = topicConf || {};

  // delete this because librdkafka will complain since this particular
  // key is a real conf value
  delete globalConf.event_cb;

  this._client = new SubClientType(globalConf, topicConf);

  // primitive callbacks from c++ wrapper
  // need binds on them if we want to broadcast the event via the emitter

  // However, the C++ wrapper lazy processes callbacks.
  // If we want to improve performance, removing this line is a good way to do it.
  // we need to do this on creation of the object
  // thats why we did some processing on config

  // Self required because we are inside an async callback function scope
  if (!no_event_cb) {
    this._client.onEvent(function eventHandler(eventType, eventData) {
      switch (eventType) {
        case 'error':
          self.emit('event.error', LibrdKafkaError.create(eventData));
          break;
        case 'stats':
          self.emit('event.stats', eventData);
          break;
        case 'log':
          self.emit('event.log', eventData);
          break;
        default:
          self.emit('event.event', eventData);
          self.emit('event.' + eventType, eventData);
      }
    });
  }

  this.metrics = {};
  this._isConnected = false;
  this.errorCounter = 0;

  /**
   * Metadata object. Starts out empty but will be filled with information after
   * the initial connect.
   *
   * @type {Client~Metadata}
   */
  this._metadata = {};

  var self = this;

  this.on('ready', function(info) {
    self.metrics.connectionOpened = Date.now();
    self.name = info.name;
  })
  .on('disconnected', function() {
    // reset metrics
    self.metrics = {};
    self._isConnected = false;
    // keep the metadata. it still may be useful
  })
  .on('event.error', function(err) {
    self.lastError = err;
    ++self.errorCounter;
  });

}
function Client ( globalConf , SubClientType , topicConf ) { if ( ! ( this instanceof Client ) ) { return new Client ( globalConf , SubClientType , topicConf ) ; } Emitter . call ( this ) ; // This superclass must be initialized with the Kafka.{Producer,Consumer} // @example var client = new Client({}, Kafka.Producer); // remember this is a superclass so this will get taken care of in // the producer and consumer main wrappers var no_event_cb = globalConf . event_cb === false ; topicConf = topicConf || { } ; // delete this because librdkafka will complain since this particular // key is a real conf value delete globalConf . event_cb ; this . _client = new SubClientType ( globalConf , topicConf ) ; // primitive callbacks from c++ wrapper // need binds on them if we want to broadcast the event via the emitter // However, the C++ wrapper lazy processes callbacks. // If we want to improve performance, removing this line is a good way to do it. // we need to do this on creation of the object // thats why we did some processing on config // Self required because we are inside an async callback function scope if ( ! no_event_cb ) { this . _client . onEvent ( function eventHandler ( eventType , eventData ) { switch ( eventType ) { case 'error' : self . emit ( 'event.error' , LibrdKafkaError . create ( eventData ) ) ; break ; case 'stats' : self . emit ( 'event.stats' , eventData ) ; break ; case 'log' : self . emit ( 'event.log' , eventData ) ; break ; default : self . emit ( 'event.event' , eventData ) ; self . emit ( 'event.' + eventType , eventData ) ; } } ) ; } this . metrics = { } ; this . _isConnected = false ; this . errorCounter = 0 ; /**
   * Metadata object. Starts out empty but will be filled with information after
   * the initial connect.
   *
   * @type {Client~Metadata}
   */ this . _metadata = { } ; var self = this ; this . on ( 'ready' , function ( info ) { self . metrics . connectionOpened = Date . now ( ) ; self . name = info . name ; } ) . on ( 'disconnected' , function ( ) { // reset metrics self . metrics = { } ; self . _isConnected = false ; // keep the metadata. it still may be useful } ) . on ( 'event.error' , function ( err ) { self . lastError = err ; ++ self . errorCounter ; } ) ; }
Base class for Consumer and Producer

This should not be created independently, but rather is
the base class on which both producer and consumer
get their common functionality.

@param {object} globalConf - Global configuration in key value pairs.
@param {function} SubClientType - The function representing the subclient
type. In C++ land this needs to be a class that inherits from Connection.
@param {object} topicConf - Topic configuration in key value pairs
@constructor
@extends Emitter
Base class for Consumer and Producer