A Kafka provider for AdonisJS v6

Adonis Kafka provides an easy way to start using Kafka.

Installation

npm i @neighbourhoodie/kafka-adonisjs

Setup

node ace configure @neighbourhoodie/kafka-adonisjs

Configuration

Edit the .env file to match your Kafka configuration.

Edit the config/kafka.js file to edit the default configuration.

Usage

Create Consumer

Create your consumer in start/kafka.js. Ex:

import Kafka from "@neighbourhoodie/adonis-kafka/services/kafka";

Kafka.on('messages', (data: any, commit: any) => {
  console.log(data)
  // commit(false) // For error transaction
  commit() // For successful transaction
});

if(Kafka.consumer) {
  Kafka.consumer.start()
}

Or create a kafka controller:

node ace make controller kafka/webhooks

// app/controllers/kafka/webhooks_controller
// import Kafka from "@neighbourhoodie/adonis-kafka/services/kafka";

export default class WebhooksController {
  async handleWebhook(data: any, commit: any) {
    console.log('received in controller', data)
    commit()
  }
}

// start/kafka.ts
import WebhooksController from "#controllers/kafka/webhooks_controller"
Kafka.on('messages', [WebhooksController, 'handleWebhook'])

if(Kafka.consumer) {
  Kafka.consumer.start()
}

Create Producer

Create your producer in app/Controllers/ for example, or in any other place. Ex:

import Kafka from "@neighbourhoodie/adonis-kafka/services/kafka";
import type { HttpContext } from '@adonisjs/core/http'

export default class UserController {
  public async show({ params }: HttpContext) {
    return Kafka.send('messages', { user_id: params.id })
  }
}

List topics

// file: start/kafka.js
import Kafka from "@neighbourhoodie/adonis-kafka/services/kafka";

Kafka.admin.listTopics().then((topics: any[]) => {
  console.log('topics', topics);
});

Create topic

// file: start/kafka.js
import Kafka from "@neighbourhoodie/adonis-kafka/services/kafka";

Kafka.admin.createTopics({
  topics: [
    {
      topic: 'messages',
      numPartitions: 1,
      replicationFactor: 1,
    },
  ],
  waitForLeaders: true,
}).then((result: any) => {
  console.log('result', result);
});

To another commands

This package uses KafkaJS, so you can use all commands from KafkaJS. Ex:

import Kafka from "@neighbourhoodie/adonis-kafka/services/kafka";

Kafka.admin.describeCluster().then((result: any) => {
  console.log('result', result);
})

Based on

• KafkaJS
• Adonis Kafka @djpfs
• Adonis Kafka @halcyon-agile