@opuscapita/kafka-clientdeprecated

OpusCapita kafka client library.

Usage no npm install needed!

<script type="module">
  import opuscapitaKafkaClient from 'https://cdn.skypack.dev/@opuscapita/kafka-client';
</script>

README

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: {
    }
}