Feature: add make:producer command

commands/make_producer.ts

@@ -0,0 +1,69 @@
+import { BaseCommand, args, flags } from '@adonisjs/core/ace'
+import string from '@adonisjs/core/helpers/string'
+import { VariableDeclarationKind } from 'ts-morph'
+import { stubsRoot } from '../stubs/main.js'
+import { CommandOptions } from '@adonisjs/core/types/ace'
+
+export default class MakeProducer extends BaseCommand {
+  static commandName = 'make:producer'
+  static description = 'Make a new Kafka producer'
+  static options: CommandOptions = {
+    allowUnknownFlags: true,
+  }
+
+  /**
+   * The name of the model file.
+   */
+  @args.string({ description: 'Kafka topic to producer data to' })
+  declare topic: string
+
+  /**
+   * Defines if we generate the factory for the model.
+   */
+  @flags.string({
+    name: 'producer',
+    alias: 'p',
+    description: 'The producer group in #start/kafka.ts',
+    default: 'default',
+  })
+  declare producer: string
+
+  /**
+   * Execute command
+   */
+  async run(): Promise<void> {
+    const codemods = await this.createCodemods()
+
+    const ProducerVariable = `${this.app.generators.modelName(this.parsed.flags.producer)}Producer`
+    const ProducerId = string.create(this.parsed.flags.producer).dashCase()
+
+    await codemods.makeUsingStub(stubsRoot, 'make/producer.stub', {
+      topic: this.topic,
+      ProducerClass: ProducerVariable,
+      entity: this.app.generators.createEntity(this.topic),
+    })
+
+    const project = await codemods.getTsMorphProject()
+    if (project) {
+      const startFile = await project.getSourceFileOrThrow(this.app.startPath('kafka.ts'))
+
+      const producerDeclaration = startFile.getVariableDeclaration(ProducerVariable)
+      if (!producerDeclaration) {
+        // adds the following to #start/kafka.ts:
+        //    export const {{ProducerVariable}} = await Kafka.createProducer('{{ProducerId}}')
+        startFile.insertVariableStatement(2, {
+          isExported: true,
+          declarationKind: VariableDeclarationKind.Const,
+          declarations: [
+            {
+              name: ProducerVariable,
+              initializer: `await Kafka.createProducer('${ProducerId}')`,
+            },
+          ],
+        })
+      }
+
+      await startFile.save()
+    }
+  }
+}

configure.ts

@@ -25,8 +25,8 @@ export async function configure(command: ConfigureCommand) {
   /**
    * Publish config file
    */
-  await codemods.makeUsingStub(stubsRoot, 'stubs/config/kafka.stub', {})
-  await codemods.makeUsingStub(stubsRoot, 'stubs/start/kafka.stub', {})
+  await codemods.makeUsingStub(stubsRoot, 'config/kafka.stub', {})
+  await codemods.makeUsingStub(stubsRoot, 'start/kafka.stub', {})
 
   /**
    * Define environment variables
@@ -69,7 +69,22 @@ export async function configure(command: ConfigureCommand) {
    * Register provider
    */
   await codemods.updateRcFile((rcFile) => {
+    rcFile.addCommand(`${command.name}/commands`)
     rcFile.addProvider(`${command.name}/kafka_provider`)
     rcFile.addPreloadFile(`#start/kafka`)
   })
+
+  /**
+   * Install packages
+   */
+  // Prompt when `install` or `--no-install` flags are not used
+  let shouldInstallPackages: boolean | undefined = command.parsedFlags.install
+  if (shouldInstallPackages === undefined) {
+    shouldInstallPackages = await command.prompt.confirm(
+      `Do you want to install additional packages required by "${command.name}"?`
+    )
+  }
+  if (shouldInstallPackages) {
+    await codemods.installPackages([{ name: 'kafkajs', isDevDependency: false }])
+  }
 }

index.ts

@@ -7,8 +7,8 @@
 |
 */
 
-export { configure } from './configure.ts'
-export { Kafka } from './src/index.ts'
-export { Consumer } from './src/consumer.ts'
-export { Producer } from './src/producer.ts'
-export { defineConfig } from './src/define_config.ts'
+export { configure } from './configure.js'
+export { Kafka } from './src/index.js'
+export { Consumer } from './src/consumer.js'
+export { Producer } from './src/producer.js'
+export { defineConfig } from './src/define_config.js'

package.json

@@ -12,12 +12,15 @@
   ],
   "exports": {
     ".": "./build/index.js",
+    "./commands": "./build/commands/main.js",
+    "./commands/*": "./build/commands/*.js",
     "./env": "./build/src/env/index.js",
     "./types": "./build/src/types.js",
     "./kafka_provider": "./build/providers/kafka_provider.js",
     "./services/kafka": "./build/services/kafka.js"
   },
   "scripts": {
+    "index:commands": "adonis-kit index build/commands",
     "clean": "del-cli build",
     "copy:templates": "copyfiles \"stubs/**/*.stub\" build",
     "typecheck": "tsc --noEmit",
@@ -27,8 +30,8 @@
     "pretest": "npm run lint",
     "test": "c8 npm run quick:test",
     "prebuild": "npm run lint && npm run clean",
-    "build": "tsup-node && tsc --emitDeclarationOnly --declaration",
-    "postbuild": "npm run copy:templates",
+    "build": "tsc",
+    "postbuild": "npm run copy:templates && npm run index:commands",
     "release": "np",
     "version": "npm run build",
     "prepublishOnly": "npm run build"
@@ -56,7 +59,6 @@
     "prettier": "^3.1.1",
     "sinon": "^17.0.1",
     "ts-node": "^10.9.2",
-    "tsup": "^8.0.2",
     "typescript": "^5.3.3"
   },
   "dependencies": {
@@ -88,21 +90,5 @@
   "eslintConfig": {
     "extends": "@adonisjs/eslint-config/package"
   },
-  "prettier": "@adonisjs/prettier-config",
-  "tsup": {
-    "entry": [
-      "./index.ts",
-      "./src/env/index.ts",
-      "./src/types.ts",
-      "./services/kafka.ts",
-      "./providers/kafka_provider.ts",
-      "./factories/main.ts"
-    ],
-    "outDir": "./build",
-    "clean": true,
-    "format": "esm",
-    "dts": false,
-    "sourcemap": true,
-    "target": "esnext"
-  }
+  "prettier": "@adonisjs/prettier-config"
 }

providers/kafka_provider.ts

@@ -1,7 +1,7 @@
 import { ApplicationService, KafkaConfig } from '@adonisjs/core/types'
 import { ContainerProviderContract } from '@adonisjs/core/types/app'
 
-import { Kafka } from '../src/index.ts'
+import { Kafka } from '../src/index.js'
 
 export default class KafkaProvider implements ContainerProviderContract {
   private app: ApplicationService

services/kafka.ts

@@ -1,5 +1,5 @@
 import app from '@adonisjs/core/services/app'
-import { Kafka } from '../src/index.ts'
+import { Kafka } from '../src/index.js'
 
 let kafka: Kafka
 

src/consumer.ts

@@ -1,13 +1,20 @@
 import { Kafka, Consumer as KafkaConsumer } from 'kafkajs'
 import { type EachMessagePayload } from 'kafkajs'
 
-import { ConsumerGroupConfig, ConsumerSubscribeTopics, ConsumerSubscribeTopic } from './types.ts'
+import type {
+  ConsumerGroupConfig,
+  ConsumerSubscribeTopics,
+  ConsumerSubscribeTopic,
+  ConsumerErrorHandler,
+  ConsumerCallback,
+  ConsumerCommitCallback,
+} from './types.ts'
 
 export class Consumer {
   config: ConsumerGroupConfig
   topics: string[]
-  events: any
-  errorHandlers: any
+  events: Record<string, ConsumerCallback[]>
+  errorHandlers: Record<string, ConsumerErrorHandler[]>
   consumer: KafkaConsumer
 
   #started: boolean = false
@@ -22,7 +29,7 @@ export class Consumer {
   }
 
   async eachMessage(payload: EachMessagePayload): Promise<void> {
-    const { topic, partition, message } = payload
+    const { topic, partition, message, heartbeat, pause } = payload
 
     let result: any
     try {
@@ -31,33 +38,49 @@ export class Consumer {
       }
       result = JSON.parse(message.value.toString())
     } catch (error) {
-      this.raiseError(topic, error)
+      this.handleError(topic, error)
       return
     }
 
-    const events = this.events[topic]
+    const callbacks = this.events[topic]
 
-    if (!events || !events.length) {
+    if (!callbacks || !callbacks.length) {
       return
     }
 
-    const promises = events.map((callback: any) => {
-      return new Promise<void>((resolve) => {
-        callback(
-          result,
-          async (commit = true) => {
-            if (this.config.autoCommit) {
-              return resolve()
-            }
-            if (commit) {
-              const offset = (Number(message.offset) + 1).toString()
-              await this.consumer.commitOffsets([{ topic, partition, offset }])
-            }
+    const promises = callbacks.map((callback) => {
+      return new Promise<void>(async (resolve, reject) => {
+        let committed = false
 
+        const committer: ConsumerCommitCallback = async (commit = true) => {
+          committed = true
+
+          if (this.config.autoCommit) {
+            return resolve()
+          }
+
+          if (commit) {
+            const offset = (Number(message.offset) + 1).toString()
+            await this.consumer.commitOffsets([{ topic, partition, offset }])
+          }
+
+          resolve()
+        }
+
+        try {
+          await callback(result, committer, heartbeat, pause)
+        } catch (error) {
+          this.handleError(topic, error)
+          resolve()
+        }
+
+        if (!committed) {
+          if (this.config.autoCommit) {
             resolve()
-          },
-          payload
-        )
+          } else {
+            reject(new Error('Expected commit() to be called as autoCommit is false'))
+          }
+        }
       })
     })
 
@@ -91,69 +114,49 @@ export class Consumer {
     return this
   }
 
-  async on({ topic, fromBeginning }: ConsumerSubscribeTopic, callback: any) {
-    const callbackFn = this.resolveCallback(callback)
-    if (!callbackFn) {
-      throw new Error('no callback specified or cannot find your controller method')
+  async on(subscription: ConsumerSubscribeTopics, callback: ConsumerCallback): Promise<void>
+  async on(subscription: ConsumerSubscribeTopic, callback: ConsumerCallback): Promise<void>
+  async on(
+    subscription: ConsumerSubscribeTopic & ConsumerSubscribeTopics,
+    callback: ConsumerCallback
+  ) {
+    if (typeof callback !== 'function') {
+      throw new TypeError('Consumer callback is not a function')
     }
 
-    if (topic instanceof RegExp) {
-      throw new Error('regexp topic not supported by adonis-kafka yet')
+    let topics = []
+    if (Array.isArray(subscription.topics)) {
+      topics = subscription.topics
+    } else {
+      topics = subscription.topic.split(',').filter((topic) => !!topic)
     }
 
-    let topicArray = [topic]
-
-    if (typeof topic === 'string') {
-      topicArray = topic.split(',')
-    }
-
-    topicArray.forEach(async (item: any) => {
-      if (!item) {
-        return
-      }
-
+    topics.forEach(async (item) => {
       const events = this.events[item] || []
 
-      events.push(callbackFn)
+      events.push(callback)
 
       this.events[item] = events
 
       this.topics.push(item)
+    })
 
-      await this.consumer.subscribe({
-        topic: item,
-        fromBeginning: fromBeginning,
-      })
+    await this.consumer.subscribe({
+      topics,
+      fromBeginning: subscription.fromBeginning ?? false,
     })
   }
 
-  raiseError(topic: string, error: Error) {
+  handleError(topic: string, error: Error) {
     const handlers = this.errorHandlers[topic] || []
-    handlers.forEach((handler: any) => {
+    handlers.forEach((handler) => {
       handler(error)
     })
   }
 
-  registerErrorHandler(topic: string, callback: any) {
-    //TODO add resolveCallback
+  onError(topic: string, callback: ConsumerErrorHandler) {
     const handlers = this.errorHandlers[topic] || []
     handlers.push(callback)
     this.errorHandlers[topic] = handlers
   }
-
-  resolveCallback(callback: any) {
-    if (Array.isArray(callback)) {
-      const [ControllerClass, fn] = callback
-      const controller = new ControllerClass()
-      if (typeof controller[fn] === 'function') {
-        return controller[fn].bind(controller)
-      }
-    }
-
-    if (typeof callback === 'function') {
-      return callback
-    }
-
-    return null
-  }
 }

src/define_config.ts

@@ -4,8 +4,6 @@ export function defineConfig(config = {}): KafkaConfig {
   return {
     brokers: 'localhost:9092',
     clientId: 'local',
-    connectionTimeout: 3000,
-    requestTimeout: 60000,
     logLevel: 'info',
     // Overwrite default config values if another one is provided
     ...config,