CDA UTILS

Installation

$ npm install

Usage of Activity Stream

const { KafkaService } = require("cda-utils");

const topics = [YOUR_KAFKA_PRODUCER];
const getProducerConfig = () => {
    let conf = {
      clientId: `your-client-id`,
      brokers: ['hostnames'],
      producerProperties: {
        transactionalId: "unique-sample-transaction-id",
        maxInFlightRequests: 1,
        idempotent: true,
      },//this config is preffred by kafkajs https://kafka.js.org/docs/transactions
    }
    if (
      process.env.KAFKA_SASL_USERNAME ||
      global.Config.broker.kafka.sasl_username
    ) {
      conf = {
        ...conf,
        "sasl.username":
          process.env.KAFKA_SASL_USERNAME ||
          global.Config.broker.kafka.sasl_username,
        "sasl.password":
          process.env.KAFKA_SASL_PASSWORD ||
          global.Config.broker.kafka.sasl_password,
        "security.protocol": "SASL_PLAINTEXT",
        "sasl.mechanisms": "PLAIN",
      };
    }
    return conf;
};

const getConsumerConfig = () => {
    let conf = {
      clientId: `your-client-id`,
      brokers: ['hostnames'],
      consumerProperties: {
        groupId: `your-group-id`,
        readUncommited: false, //only for producer using transactions
      },
    }
    if (
      process.env.KAFKA_SASL_USERNAME ||
      global.Config.broker.kafka.sasl_username
    ) {
      conf = {
        ...conf,
        "sasl.username":
          process.env.KAFKA_SASL_USERNAME ||
          global.Config.broker.kafka.sasl_username,
        "sasl.password":
          process.env.KAFKA_SASL_PASSWORD ||
          global.Config.broker.kafka.sasl_password,
        "security.protocol": "SASL_PLAINTEXT",
        "sasl.mechanisms": "PLAIN",
      };
    }
    return conf;
};

const Kafka = new KafkaService(getProducerConfig());
const kafkaEventDelegator = async (messageReceived) => {
  try {
    const message = messageReceived.content;
    if (message.eventName === "test") {
      console.log("test kafka event");
      return false;   //this will mark message as processed
    }
  } catch (e) {
      return false;   //this will push the message for DLQ
      console.log(error.response.data);
  }

  // Do something with the message
};
const kafkaConsumerConfig = {
  prefetch: 10,
  ...getConsumerConfig(),
};
let index= 0;
async function testPushToQueue() {
  await Kafka.pushToQueue(
    "dev-test",
    [
      {
        eventName: "test",
        data: {
          val: `hello world ${index} `,
          index: 2,
        },
      },
    ],
    {
      "replication-factor": 1,
    }
  );
  await Kafka.commitTransaction(); //commit or abort this so transaction can reach the end state
  index++;
}
async function init() {
  Kafka.listenQueue(topics, kafkaEventDelegator, kafkaConsumerConfig);
  setInterval(testPushToQueue, 10000);
}
try {
  init();
} catch (error) {
  console.log(error);
}

Usage of Kafka Broker

const cda_utils = require('cda-utils');
const Kafka_Broker = cda_utils.Broker.Kafka;

Mandatory Config for cda_utils

global.Config = {};
global.Config.env = '';
global.Config.cda_utils = {};
global.Config.cda_utils.queueName = 'skuad-events';
global.Config.cda_utils.broker = {};
global.Config.cda_utils.broker.maxQueueSize = '100'
global.Config.cda_utils.broker.kafka = {
    brokers: 'localhost:9092',
    commitTimeInterval: '1',
    maxParallelHandles: '10',
    messageProcessingTimeoutMS: '1'
};
};