Feat/kafka-setting: 트랜잭션 락 최소화를 위한 성능 개선

.gitignore

@@ -25,3 +25,5 @@ turbo-build.log
 /playwright-report/
 /blob-report/
 /playwright/.cache/
+
+.windsurfrules

app/koi-server/docker/docker-compose.yml

@@ -0,0 +1,37 @@
+services:
+  zookeeper:
+    image: bitnami/zookeeper:latest
+    container_name: zookeeper
+    ports:
+      - 2181:2181
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+
+  kafka:
+    image: bitnami/kafka:latest
+    container_name: kafka
+    depends_on:
+      - zookeeper
+    ports:
+      - 9092:9092
+      - 9094:9094
+    environment:
+      - ALLOW_PLAINTEXT_LISTENER=yes
+      - KAFKA_ENABLE_KRAFT=no
+      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
+      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
+      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
+      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,EXTERNAL://localhost:9094
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT
+
+  kafka-ui:
+    image: provectuslabs/kafka-ui:latest
+    container_name: kafka-ui
+    depends_on:
+      - kafka
+    ports:
+      - 8080:8080
+    environment:
+      - DYNAMIC_CONFIG_ENABLED=true
+      - KAFKA_CLUSTERS_0_NAME=koi_kafka
+      - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092

app/koi-server/jest.config.js

@@ -0,0 +1,11 @@
+module.exports = {
+  collectCoverageFrom: ['**/*.(t|j)s'],
+  coverageDirectory: '../coverage',
+  moduleFileExtensions: ['js', 'json', 'ts'],
+  rootDir: 'src',
+  testEnvironment: 'node',
+  testRegex: '.*\\.spec\\.ts$',
+  transform: {
+    '^.+\\.(t|j)s$': 'ts-jest',
+  },
+};

app/koi-server/package.json

@@ -10,18 +10,20 @@
     "start:debug": "nest start --debug --watch",
     "start:prod": "node dist/main",
     "lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
-    "test": "jest",
-    "test:watch": "jest --watch",
-    "test:cov": "jest --coverage",
-    "test:debug": "node --inspect-brk -r tsconfig-paths/register -r ts-node/register node_modules/.bin/jest --runInBand",
+    "test": "jest --config jest.config.js",
+    "test:watch": "jest --config jest.config.js --watch",
+    "test:cov": "jest --config jest.config.js --coverage",
     "test:e2e": "jest --config ./test/jest-e2e.json",
+    "test:debug": "node --inspect-brk -r tsconfig-paths/register -r ts-node/register node_modules/.bin/jest --runInBand",
     "turbo:dev": "turbo run dev"
   },
   "dependencies": {
     "@codegenie/serverless-express": "^4.13.0",
+    "@confluentinc/kafka-javascript": "^1.2.0",
     "@nestjs/common": "^10.3.0",
     "@nestjs/config": "^3.1.1",
     "@nestjs/core": "^10.3.0",
+    "@nestjs/microservices": "^11.0.7",
     "@nestjs/mongoose": "^10.0.2",
     "@nestjs/platform-express": "^10.3.0",
     "@toss/date": "^1.1.7",
@@ -33,6 +35,7 @@
     "express": "^4.18.2",
     "feature-nest-poll": "workspace:*",
     "feature-nest-stock": "workspace:*",
+    "kafkajs": "^2.2.4",
     "mongoose": "^8.1.0",
     "reflect-metadata": "^0.1.13",
     "rxjs": "^7.2.0"

app/koi-server/src/app/app.controller.ts

@@ -1,7 +1,10 @@
-import { Controller, Post, Query } from '@nestjs/common';
+import { Body, Controller, Get, Param, Post, Query } from '@nestjs/common';
+import { AppService } from './app.service';
 
 @Controller()
 export class AppController {
+  constructor(private readonly appService: AppService) {}
+
   @Post('log')
   log(@Query('msg') msg: string): boolean {
     fetch(
@@ -19,4 +22,24 @@ export class AppController {
     console.warn(`${decodeURIComponent(msg)} (${new Date()})`);
     return true;
   }
+
+  @Get()
+  getHello(): string {
+    return this.appService.getHello();
+  }
+
+  @Post('add-topic')
+  async addSubscriptionTopic(@Body('topic') topic: string): Promise<string> {
+    console.log(topic);
+    if (topic === undefined) {
+      return 'topic is undefined';
+    }
+    await this.appService.addSubscriptionTopic(topic);
+    return `topic ${topic} added`;
+  }
+
+  @Post('send/:topic')
+  async sendMessage(@Body() msg: { value: string }, @Param('topic') topic: string): Promise<void> {
+    await this.appService.sendMessage(topic, msg.value);
+  }
 }

app/koi-server/src/app/app.service.ts

@@ -1,6 +1,49 @@
 import { Injectable } from '@nestjs/common';
+import { EachMessagePayload, Kafka } from 'kafkajs';
 
 @Injectable()
 export class AppService {
-  constructor() {}
+  private kafka = new Kafka({
+    brokers: ['localhost:9094'],
+    clientId: 'koi_kafka',
+  });
+
+  private producer = this.kafka.producer();
+
+  private consumer = this.kafka.consumer({ groupId: 'koi_kafka_group' });
+
+  constructor() {
+    this.consumer.connect(); // 접속
+    this.consumer.subscribe({ topics: ['test_a', 'test_b'] }); // 구독
+    this.consumer.run({
+      eachMessage: this.consumerCallback, // 메세지 수신 콜백
+    });
+  }
+
+  async consumerCallback(payload: EachMessagePayload): Promise<void> {
+    // 메세지 수신 콜백
+    console.log('kafka message arrived');
+    console.log(`topic: ${payload.topic}, Message:${payload.message.value.toString()}`);
+  }
+
+  async addSubscriptionTopic(topic: string): Promise<void> {
+    await this.consumer.stop(); // 컨슈머 멈추고
+    await this.consumer.subscribe({ topic }); // 구독하고
+    await this.consumer.run({
+      eachMessage: this.consumerCallback,
+    });
+  }
+
+  async sendMessage(topic: string, message: string): Promise<void> {
+    await this.producer.connect();
+    await this.producer.send({
+      messages: [{ value: message }],
+      topic,
+    });
+    await this.producer.disconnect();
+  }
+
+  getHello(): string {
+    return 'Hello World!';
+  }
 }

app/koi-server/test/app.e2e-spec.ts

@@ -1,7 +1,7 @@
 import { Test, TestingModule } from '@nestjs/testing';
 import { INestApplication } from '@nestjs/common';
-import * as request from 'supertest';
-import { AppModule } from '../src/app.module';
+import request from 'supertest';
+import { AppModule } from '../src/app/app.module';
 
 describe('AppController (e2e)', () => {
   let app: INestApplication;