busie-node-kafka

A node module that provides an api for easily creating and interacting with Kafka Producers and Consumers.

Usage

Producer

import { Producer } from 'busie-node-kafka';
const config = {
  // Kafka Config here
  // See node-rdkafka docs for valid config
};
const onDelivery = (err, record) => {
  // some callback function to be invoked on message delivery
  // see node-rdkafka docs for usage.
};
const type = 'type'; // some arbitrary type to assign to the producer. used for error messaging and tracing.
const producer = new Producer(type, config, onDelivery);

Producing a keyed message

// producer initialization omitted for brevity
producer.produceKeyedMessage({ topic, partition, message, key, timestamp, opaque });

Note the named arguments. These arguments are required.

Producing a message

// producer initialization omitted for brevity
producer.produceMessage({ topic, partition, message });

Note the named arguments. These arguments are required

Destroying the producer

//producer initialization omitted for brevity
Producer.destroy(producer);

Consumer

import { Consumer } from 'busie-node-kafka';
const config = {
  // Kafka config here
  // see node-rdkafka docs valid config
};
const offsetConfig = {
  // KafkaConsumer offset config here
  // see node-rdkafka docs for valid offset config
};
const onData = () => {
  // some callback to be invoked on data consumption
  // see node-rdkafka for usage
};
const type = 'type'; // some arbitrary type to assign to the consumer. used for error messaging and tracing.
const consumer = new Consumer(type, config, offsetConfig, onData);

Subscribing to topics

// consumer initialization omitted for brevity
consumer.subscribe(['topic1', 'topic2']);

Consuming events

// consumer initialization omitted for brevity
// topic subscription omitted for brevity
consumer.consume();

Destroying a consumer

// consumer initialization omitted for brevity
Consumer.destroy(consumer);

Kafka

This module provides access to the base Kafka object that comes with node-rdkafka

import { Kafka } from 'busie-node-kafka';
// Access to Kafka.Producer, Kafka.KafkaConsumer, Kafka.AdminClient etc. now available