Adding back platformatic

Author: CarlosGamero

packages/kafka/lib/test.spec.ts

@@ -1,6 +1,14 @@
 import { randomUUID } from 'node:crypto'
 import { KafkaJS } from '@confluentinc/kafka-javascript'
 import { waitAndRetry } from '@lokalise/universal-ts-utils/node'
+import {
+  Consumer,
+  type Message,
+  ProduceAcks,
+  Producer,
+  stringDeserializers,
+} from '@platformatic/kafka'
+import { stringSerializers } from '@platformatic/kafka'
 import { type TestContext, registerDependencies } from '../test/testContext.ts'
 
 describe('Test', () => {
@@ -14,59 +22,112 @@ describe('Test', () => {
     await testContext.dispose()
   })
 
-  it('should send and receive a message', async () => {
-    // Given
-    const clientId = randomUUID()
-    const groupId = randomUUID()
-    // Use a fresh, unique topic per run to avoid stale state
-    const topic = `admin-test-topic-${Date.now()}`
-    const messageValue = 'My test message'
-
-    const kafka = new KafkaJS.Kafka({
-      'client.id': clientId,
-      'bootstrap.servers': testContext.cradle.kafkaConfig.brokers.join(','),
-    })
+  describe('confluentic', () => {
+    it('should send and receive a message', async () => {
+      // Given
+      const clientId = randomUUID()
+      const groupId = randomUUID()
+      // Use a fresh, unique topic per run to avoid stale state
+      const topic = `test-topic-${Date.now()}`
+      const messageValue = 'My test message'
 
-    // Topics can be created from producers, but as we will first connect a consumer, we need to create the topic first
-    const admin = kafka.admin()
-    await admin.connect()
-    await admin.createTopics({
-      topics: [{ topic }],
-    })
+      const kafka = new KafkaJS.Kafka({
+        'client.id': clientId,
+        'bootstrap.servers': testContext.cradle.kafkaConfig.brokers.join(','),
+      })
 
-    const messages: string[] = []
+      // Topics can be created from producers, but as we will first connect a consumer, we need to create the topic first
+      const admin = kafka.admin()
+      await admin.connect()
+      await admin.createTopics({
+        topics: [{ topic }],
+      })
 
-    const consumer = kafka.consumer({ 'group.id': groupId })
-    await consumer.connect()
-    await consumer.subscribe({ topic })
+      const messages: string[] = []
 
-    await consumer.run({
-      eachMessage: ({ message }) => {
-        const messageString = message.value?.toString()
-        if (messageString) messages.push(messageString)
-        return Promise.resolve()
-      },
-    })
-    // Wait for the consumer to be assigned partitions
-    await waitAndRetry(() => consumer.assignment().length > 0, 100, 10)
-
-    // When
-    const producer = kafka.producer()
-    await producer.connect()
-    await producer.send({
-      topic,
-      messages: [{ value: messageValue }],
+      const consumer = kafka.consumer({ 'group.id': groupId })
+      await consumer.connect()
+      await consumer.subscribe({ topic })
+
+      await consumer.run({
+        eachMessage: ({ message }) => {
+          const messageString = message.value?.toString()
+          if (messageString) messages.push(messageString)
+          return Promise.resolve()
+        },
+      })
+      // Wait for the consumer to be assigned partitions
+      await waitAndRetry(() => consumer.assignment().length > 0, 100, 10)
+
+      // When
+      const producer = kafka.producer()
+      await producer.connect()
+      await producer.send({
+        topic,
+        messages: [{ value: messageValue }],
+      })
+
+      // Then
+      await waitAndRetry(() => messages.length > 0)
+
+      // Cleaning up before checks to avoid stale state
+      await consumer.disconnect()
+      await producer.disconnect()
+      await admin.disconnect()
+
+      expect(messages).toHaveLength(1)
+      expect(messages[0]).toEqual(messageValue)
     })
+  })
+
+  describe('platformatic', () => {
+    it('should send and receive a message', async () => {
+      // Given
+      const clientId = randomUUID()
+      // Use a fresh, unique topic per run to avoid stale state
+      const topic = `test-topic-${Date.now()}`
+      const messageValue = 'My test message'
+
+      const receivedMessages: Message<string, string, string, string>[] = []
+
+      // Create producer
+      const producer = new Producer({
+        clientId,
+        bootstrapBrokers: testContext.cradle.kafkaConfig.brokers,
+        serializers: stringSerializers,
+        autocreateTopics: true,
+      })
 
-    // Then
-    await waitAndRetry(() => messages.length > 0)
+      // Create consumer
+      const consumer = new Consumer({
+        clientId,
+        groupId: randomUUID(),
+        bootstrapBrokers: testContext.cradle.kafkaConfig.brokers,
+        deserializers: stringDeserializers,
+        autocreateTopics: true,
+      })
 
-    // Cleaning up before checks to avoid stale state
-    await consumer.disconnect()
-    await producer.disconnect()
-    await admin.disconnect()
+      const stream = await consumer.consume({ topics: [topic] })
+      stream.on('data', (message) => {
+        receivedMessages.push(message)
+        stream.close()
+      })
 
-    expect(messages).toHaveLength(1)
-    expect(messages[0]).toEqual(messageValue)
+      // When
+      await producer.send({
+        messages: [{ topic, value: messageValue }],
+        acks: ProduceAcks.NO_RESPONSE,
+      })
+
+      // Then
+      await waitAndRetry(() => receivedMessages.length > 0)
+
+      // Cleanup
+      producer.close()
+      consumer.close()
+
+      expect(receivedMessages).toHaveLength(1)
+      expect(receivedMessages[0]?.value?.toString()).toBe(messageValue)
+    })
   })
 })

packages/kafka/package.json

@@ -50,6 +50,7 @@
         "@confluentinc/kafka-javascript": "^1.3.0",
         "@lokalise/node-core": "^14.0.1",
         "@lokalise/universal-ts-utils": "^4.4.1",
+        "@platformatic/kafka": "^1.2.0",
         "zod": "^3.25.7"
     },
     "peerDependencies": {