A Kafka provider for AdonisJS v6
Adonis Kafka provides an easy way to start using Kafka.
npm i @neighbourhoodie/kafka-adonisjsnode ace configure @neighbourhoodie/kafka-adonisjsEdit the .env file to match your Kafka configuration.
Edit the config/kafka.js file to edit the default configuration.
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 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 })
}
}// file: start/kafka.js
import Kafka from "@neighbourhoodie/adonis-kafka/services/kafka";
Kafka.admin.listTopics().then((topics: any[]) => {
console.log('topics', topics);
});// 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);
});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);
})