lib/baseProducer.js
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