function BaseProducer (client, options, defaultPartitionerType, customPartitioner) {
  EventEmitter.call(this);
  options = options || {};

  this.ready = false;
  this.client = client;

  this.requireAcks = options.requireAcks === undefined ? DEFAULTS.requireAcks : options.requireAcks;
  this.ackTimeoutMs = options.ackTimeoutMs === undefined ? DEFAULTS.ackTimeoutMs : options.ackTimeoutMs;

  if (customPartitioner !== undefined && options.partitionerType !== PARTITIONER_TYPES.custom) {
    throw new Error('Partitioner Type must be custom if providing a customPartitioner.');
  } else if (customPartitioner === undefined && options.partitionerType === PARTITIONER_TYPES.custom) {
    throw new Error('No customer partitioner defined');
  }

  var partitionerType = PARTITIONER_MAP[options.partitionerType] || PARTITIONER_MAP[defaultPartitionerType];

  // eslint-disable-next-line
  this.partitioner = new partitionerType(customPartitioner);

  this.connect();
}
function BaseProducer (client, options, defaultPartitionerType, customPartitioner) {
  EventEmitter.call(this);
  options = options || {};

  this.ready = false;
  this.client = client;

  this.requireAcks = options.requireAcks === undefined ? DEFAULTS.requireAcks : options.requireAcks;
  this.ackTimeoutMs = options.ackTimeoutMs === undefined ? DEFAULTS.ackTimeoutMs : options.ackTimeoutMs;

  if (customPartitioner !== undefined && options.partitionerType !== PARTITIONER_TYPES.custom) {
    throw new Error('Partitioner Type must be custom if providing a customPartitioner.');
  } else if (customPartitioner === undefined && options.partitionerType === PARTITIONER_TYPES.custom) {
    throw new Error('No customer partitioner defined');
  }

  var partitionerType = PARTITIONER_MAP[options.partitionerType] || PARTITIONER_MAP[defaultPartitionerType];

  // eslint-disable-next-line
  this.partitioner = new partitionerType(customPartitioner);

  this.connect();
}
function BaseProducer ( client , options , defaultPartitionerType , customPartitioner ) { EventEmitter . call ( this ) ; options = options || { } ; this . ready = false ; this . client = client ; this . requireAcks = options . requireAcks === undefined ? DEFAULTS . requireAcks : options . requireAcks ; this . ackTimeoutMs = options . ackTimeoutMs === undefined ? DEFAULTS . ackTimeoutMs : options . ackTimeoutMs ; if ( customPartitioner !== undefined && options . partitionerType !== PARTITIONER_TYPES . custom ) { throw new Error ( 'Partitioner Type must be custom if providing a customPartitioner.' ) ; } else if ( customPartitioner === undefined && options . partitionerType === PARTITIONER_TYPES . custom ) { throw new Error ( 'No customer partitioner defined' ) ; } var partitionerType = PARTITIONER_MAP [ options . partitionerType ] || PARTITIONER_MAP [ defaultPartitionerType ] ; // eslint-disable-next-line this . partitioner = new partitionerType ( customPartitioner ) ; this . connect ( ) ; }
Provides common functionality for a kafka producer

@param {Client} client A kafka client object to use for the producer
@param {Object} [options] An object containing configuration options
@param {Number} [options.requireAcks=1] Configuration for when to consider a message as acknowledged.
<li>0 = No ack required</li>
<li>1 = Leader ack required</li>
<li>-1 = All in sync replicas ack required</li>

@param {Number} [options.ackTimeoutMs=100] The amount of time in milliseconds to wait for all acks before considered
the message as errored
@param {Number} [defaultPartitionType] The default partitioner type
@param {Object} [customPartitioner] a custom partitinoer to use of the form: function (partitions, key)
@constructor
Provides common functionality for a kafka producer