@invisible/kafka

Usage

import { makeKafkaInstance, TOPICS } from '@invisible/kafka'

const { consumerFactory, producerFactory, start, stop } = makeKafkaInstance({
  clientId: 'testy',
  brokers: ['localhost:9092'],
})

consumerFactory({
  topic: TOPICS.INSTANCE_UPDATED,
  runEachMessage: async (args: EachMessagePayload) => {
    console.log(`Instance ${message.value} just got updated.`)
  }
})

const { producer } = producerFactory()

(async () => {
  await start()
  await producer.send({ topic: TOPICS.INSTANCE_UPDATED, messages: [ { value: Math.round(Math.random()*100) } ] })
  await stop()
})()

Tips

Make sure to also set up safe-stop events:

process
  .on('exit',  stop)
  .on('SIGINT', stop)
  .on('uncaughtException', stop)
  .on('unhandledRejection', stop)