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";
const consumer = Kafka.createConsumer({ groupId: 'default'})
consumer.on({ topic: 'messages' }, (data: any, commit: any) => {
console.log(data)
// commit(false) // For error transaction
commit() // For successful transaction
});
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'
const consumer = Kafka.createConsumer({ groupId: 'default' })
consumer.on({ topic: 'messages' }, [WebhooksController, 'handleWebhook'])
consumer.start()const consumer = Kafka.createConsumer({ groupId: 'default' })
consumer.on('messages', [WebhooksController, 'handleWebhook'])
consumer.registerErrorHandler('messsages', (error) => {
console.log('got error!', error)
})
consumer.start()At the moment, only JSON.parse() errors are handled.
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 {
constructor() {
Kafka.createProducer('myProducer', {} /* ProducerConfig */).start()
}
public async show({ params, kafka: { producers } }: HttpContext) {
return producers['myProducer'].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);
})