kafka node consumer

example.js

const kafkaConsumer = require('./kafka-consumer.js');

kafkaConsumer.observableFor('support_calls').subscribe( (data) => dostuff.with(data) );

kafka-config.json

{
  "kafkaHost": "10.194.91.83:9092",
  "groupId": "whatever",
  "sessionTimeout": 15000
}

kafka-consumer.js

const kafka = require('kafka-node');
const kafkaConfig = require('./kafka-config.json');
const { fromEvent } = require('rxjs');
const { map } = require("rxjs/operators");

const options = {
  ...kafkaConfig,
  onRebalance: (isAlreadyMember, callback) => { callback(); } // or null
};
 
observableFor = (topic) => {
  const consumerGroup = new kafka.ConsumerGroup(options, [topic]);

  return fromEvent(consumerGroup, "message").pipe(
    map(item => JSON.parse(JSON.parse(item.value))),
    map(invertObject),
    map(arrayify)
  );
}

module.exports = { observableFor };

function invertObject(inputObject) {
  const outputObject = {};
  Object.keys(inputObject).forEach((innerKey) => {
    Object.keys(inputObject[innerKey]).forEach((outerKey) => {
      if(!outputObject[outerKey]){
        outputObject[outerKey] = {};
      }
      outputObject[outerKey][innerKey] = inputObject[innerKey][outerKey];
    });
  });
  return outputObject;
}

function arrayify(inputObject) {
  return Object.keys(inputObject).map( id => ({ id, ...inputObject[id]}));
}