Method commit

lib/commitStream.js:125–152

(cb)

  }

  commit (cb) {
    let self = this;

    if (!cb) {
      cb = function noop () {};
    }

    if (self.committing) {
      return cb(null, 'Commit in progress');
    }

    let topicPartionOffsets = self.topicPartionOffsets;

    let commits = topicPartionOffsets.filter(function (partition) {
      return partition.offset !== 0;
    });

    if (commits.length) {
      self.committing = true;
      self.client.sendOffsetCommitRequest(self.groupId, commits, function () {
        self.emit('commitComplete', { group: self.groupId, commits });
        self.committing = false;
        cb.apply(this, arguments);
      });
    } else {
      cb(null, 'Nothing to be committed');
    }
  }
}

module.exports = CommitStream;