wip

service/dev-external-deps/docker-compose.yaml

@@ -21,3 +21,32 @@ services:
       POSTGRES_DB: terraapprove
     ports:
       - 5433:5432
+
+  message-broker:
+    container_name: message-broker
+    image: bitnami/kafka:latest
+    ports:
+      - "9092:9092"
+      - "9093:9093"
+      - "9997:9997"
+    environment:
+      KAFKA_CFG_NODE_ID: 0
+      KAFKA_CFG_PROCESS_ROLES: controller,broker
+      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
+      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://message-broker:9092
+      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 0@message-broker:9093
+      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_CFG_KAFKA_JMX_PORT: 9997
+      KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=message-broker -Dcom.sun.management.jmxremote.rmi.port=9997
+      KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "true"
+
+  message-broker-ui:
+    container_name: message-broker-ui
+    image: provectuslabs/kafka-ui
+    ports:
+      - "8080:8080"
+    environment:
+      KAFKA_CLUSTERS_0_NAME: dev
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: message-broker:9092
+      KAFKA_CLUSTERS_0_METRICS_PORT: 9997

service/main/src/app.module.ts

@@ -2,10 +2,11 @@ import {Module} from "@nestjs/common"
 import {ServiceModule} from "@libs/service/service.module"
 import {SourceCodeController} from "./controller/source-code.controller"
 import {PlanController} from "./controller/plan.controller"
+import {KafkaController} from "./controller/KafkaController"
 
 @Module({
   imports: [ServiceModule],
-  controllers: [SourceCodeController, PlanController],
+  controllers: [SourceCodeController, PlanController, KafkaController],
   providers: []
 })
 export class AppModule {}

service/main/src/controller/KafkaController.ts

@@ -0,0 +1,66 @@
+import {Controller, Get, Injectable, Logger, Post} from "@nestjs/common"
+import {randomUUID} from "crypto"
+import {Kafka, Partitioners} from "kafkajs"
+
+@Controller()
+export class KafkaController {
+  constructor() {
+    console.log("KafkaController constructor")
+  }
+
+  @Post("kafka")
+  async postKafka() {
+    const kafka = new Kafka({
+      clientId: "my-app",
+      brokers: ["localhost:9092"],
+      retry: {
+        retries: 1
+      }
+    })
+
+    const producer = kafka.producer({
+      createPartitioner: Partitioners.DefaultPartitioner
+    })
+
+    await producer.connect()
+    await producer.send({
+      topic: "test-3",
+      messages: [{value: `Hello KafkaJS user! ${randomUUID()}`}]
+    })
+
+    return ""
+  }
+
+  @Get("kafka")
+  async getKafka() {
+    const kafka = new Kafka({
+      clientId: "my-app",
+      brokers: ["localhost:9092"],
+      retry: {
+        retries: 1
+      }
+    })
+
+    const consumer = kafka.consumer({
+      groupId: randomUUID()
+    })
+
+    await consumer.connect()
+    await consumer.subscribe({topic: "test-3", fromBeginning: true})
+
+await consumer.
+
+    await consumer.run({
+      eachMessage: async ({topic, partition, message}) => {
+        Logger.log({
+          topic,
+          partition,
+          offset: message.offset,
+          value: message.value!!.toString()
+        })
+      }
+    })
+
+    return ""
+  }
+}

service/package.json

@@ -35,6 +35,7 @@
     "@prisma/client": "5.9.0",
     "express": "4.18.2",
     "fp-ts": "2.16.2",
+    "kafkajs": "2.2.4",
     "reflect-metadata": "0.2.1",
     "rxjs": "7.8.1"
   },

service/yarn.lock

@@ -4976,6 +4976,13 @@ __metadata:
   languageName: node
   linkType: hard
 
+"kafkajs@npm:2.2.4":
+  version: 2.2.4
+  resolution: "kafkajs@npm:2.2.4"
+  checksum: 75eb0d221397085f90e51f8a2d752495c9fa9c1b3a1a6db610cd7074fa8c52777f295832fd0a7c49cded5e574337a09fafa8c3f7cf1caa38f4dc9aa20fcfb7df
+  languageName: node
+  linkType: hard
+
 "keyv@npm:^4.5.3":
   version: 4.5.4
   resolution: "keyv@npm:4.5.4"
@@ -6723,6 +6730,7 @@ __metadata:
     express: "npm:4.18.2"
     fp-ts: "npm:2.16.2"
     jest: "npm:29.7.0"
+    kafkajs: "npm:2.2.4"
     openapi-typescript: "npm:6.7.4"
     prettier: "npm:2.8.8"
     prisma: "npm:5.9.0"