@imec/digital-twin-kafka-utils

Kafka Connection Manager

Description

Kafka manager is a wrapper around Kafkajs for the Digital Twin project. It's meant to be available publicly from NPM.

How to update this package

First of all you need to be a collaborator of the npm package. Ask [e-mail] or [e-mail] to get access.

1. Create a branch
2. Update code
3. Run yarn build
4. Run run npm version <patch,minor,major>
5. Commit changes and the version update
6. Open PR
7. Merge PR
8. Checkout to master
9. Run npm publish on master

How to use it

Example to create and initialize an instance

import KafkaManager, { ProducerRecord, KafkaMessage } from ".";

interface RoadData{
	name: string,
	age: number
}

interface HobbitData{
	name: string,
	address: string
}

export default class Example {
  kafka!: KafkaManager

	async init() {
    this.kafka = new KafkaManager({ brokers: ['kafka-server:9092']});
    const producerConfig = {} // producer config from kafkajs, optional can be left out
    const kafkaProducer = await this.kafka.createProducer(producerConfig);
    const consumerConfig = {groupId: "hello"};
    const kafkaConsumer = await this.kafka.createConsumer(consumerConfig);
    const topic = 'traffic-loop';

		const record: ProducerRecord = {
			topic,
			messages: [
        {
          headers: {'sequenceId': '2'},
          key: 'remove road',
          value: {name: 'my awesome name', age: 124}
        },
        {
          headers: {'sequenceId': '2'},
          key: 'add hobbit',
          value: {name: 'hobbit', address: 'Esgaroth'}
        }
      ]
	  }

    setInterval(() => {
      console.log('publishing');
      this.kafka.publish(kafkaProducer, record);
    }, 5000)

    function eachBatchFn(messages: KafkaMessage<RoadData|HobbitData>[]) {
      messages.forEach(element => {
        switch (element.key) {
        case "add hobbit":
          console.log("Handle the hobbit message");
          break;
        case "remove road":
          console.log("Handle the road message");
          break;
        default:
          console.log("Handle unexpected default");
        }
      });
      console.log("batch messages", messages)
    }

    function eachMessageFn(message: KafkaMessage<RoadData|HobbitData>) {
      console.log("messages", message)
	    switch (message.key) {
        case "add hobbit":
			    console.log("Handle the hobbit message");
          break;
        case "remove road":
			    console.log("Handle the road message");
          break;
        default:
          console.log("Handle unexpected default");
        }
		}

    // Subscribing to a topic can be done using the two methods below. Note that you cannot have `eachMessage` and `eachBatch` in the same subscription! This will cause kafkajs to only run one of the two.

    // Use for each message subscription
    this.kafka.subscribe<RoadData|HobbitData>(kafkaConsumer, {
      subscribeTopic: {topic},
      eachMessage: eachMessageFn
    });

    // Use batch subscription
    this.kafka.subscribe<RoadData|HobbitData>(kafkaConsumer, {
      subscribeTopic: {topic},
      eachBatch: eachBatchFn,
    });
	}
}