DEPRECATED

You should probably NOT be using this library for the following purposes:

• not much value is added over kafkajs
• many interfaces of native kafkajs are hidden
• assumption of autocommit
  • every message processed with a separate call (no batch processing)
  • autocommit could lead to data loss
• not used in OpusCapita

However you might find it usable as there is a lot of logic implemented and it is using TypeScript. In such a case feel free to unarchive it or fork.

This module implements a high-level Kafka consumer/producer. It is based on kafkajs and adds some convenience functions on top of the provided functionallity:

• Supports dead letter queing by implementing a simplified, one-stage DLQ without retry.
• Enveloping: Every message is enriched by platform specific information (context) that is transparent to the application using kafka-client
• Disables usage of wildcard topics
• On-demand subscribe and unsubscribe
• (tba) Metrics: Implement metrics for active health monitoring of client connections

Please find more information on the individual topics in the wiki.

Minimum setup

First got to your local code directory and run:

npm install @opuscapita/kafka-client

Kafka-client uses consul for service discovery meaning you need to have access to a running Consul server to get your endpoint configuration for the broker. In addition, the Kafka broker itself is required which also has to be registered inside Consul.

The basic implementation requirements defines the naming conventions used for topics as follows: serviceName.domainName.eventType.eventDetail.

Common subscription

If all this is set up, a new client connection can be created like this:

const KafkaClient = require('@opuscapita/kafka-client');
const kafka = new KafkaClient();
await kafka.init();

Consuming and producing works analogous to the well-known rabbitMq-based event-client library. Attention: Wildcard subscriptions are no longer supported.

(async () =>
{
    // Subscribe to a channel by name.
    await events.subscribe('my-service.my-channel', (m ) => { console.log(m); return true; });
    await events.emit('my-service.my-channel', 'Hello, world!');

    // unsubscribe from a particular key
    await events.unsubscribe('my-service.my-channel');

    // gracefully shutdown
    await events.dispose();
})();

If the application callback provided as the second parameter provided to the subscribe() method, the message handling will be considered errornous and the message will be put the DLQ.

Fetching single message

tbi

Default configuration

The default configuration object provides hints about what the module's standard behavior is like.

{
    serializer: JSON.stringify,
    parser: JSON.parse,
    serializerContentType: 'application/json',
    parserContentType: 'application/json',
    consumerGroupId: null,
    enableHealthChecks: true,
    logger: new Logger(),
    consul: {
        host: 'consul',
        mqServiceName: 'kafka'
    },
    consulOverride: {
        host: null
    },
    context: {
    }
}