@imec/digital-twin-kafka-utils

Kafka Connection Manager

Description

Styleguide used for the Digital Twin project. It's meant to be available publicly from NPM.

How to use it

You need to make sure to initialize the KafkaConnectionManager instance before using the producer or consumer.

Example to create and initialize an instance

import createKafkaManager from 'digital-twin-kafka-utils'

const kafka = await createKafkaManager({
  ip: 'localhost',
  port: '9092',
  clientId: 'digital-twin',
  consumerGroupId: 'my-group'
});

Example on how to publish a message

Send a single or multiple messages on a topic. The payload is mapped to the value of the kafka message. It's used as an array so we can send a batch if needed.

await kafka.publish({
    topic: 'my-topic',
    type: 'Road Closing',
    headers: {},
    payload: [`{}`],
    compression: CompressionTypes.GZIP,
  })

Example on how to subscribe to a topic

/!\ If using eachMessage, the eachBatch won't be called as it's the way kafkaJS has done the implementation.

// By each message
await kafka.subscribe({
    topic: 'my-topic',
    eachMessage: (messagePayload) => {
      console.log("messagePayload", messagePayload);
    },
})
// By batch
await kafka.subscribe({
    topic: 'my-topic',
    eachBatch: (batchPayload) => {
      console.log("batchPayload", batchPayload);
    },
})