Merge pull request #478 from boostcampwm-2024/feat/rabbitmq-messaging-layer

✨Feat: RabbitMQ 클라이언트 레이어 구현

Author: CodeVac513

docker-compose/definitions.json

@@ -0,0 +1,86 @@
+{
+  "exchanges": [
+    {
+      "name": "email",
+      "vhost": "/",
+      "type": "direct",
+      "durable": true
+    },
+    {
+      "name": "crawling",
+      "vhost": "/",
+      "type": "topic",
+      "durable": true
+    },
+    {
+      "name": "dead_letter",
+      "vhost": "/",
+      "type": "topic",
+      "durable": true
+    }
+  ],
+  "queues": [
+    {
+      "name": "email_send",
+      "vhost": "/",
+      "durable": true,
+      "arguments": {
+        "x-dead-letter-exchange": "dead_letter",
+        "x-dead-letter-routing-key": "email.dead_letter"
+      }
+    },
+    {
+      "name": "crawling_full",
+      "vhost": "/",
+      "durable": true,
+      "arguments": {
+        "x-dead-letter-exchange": "dead_letter",
+        "x-dead-letter-routing-key": "crawling.full.dead_letter"
+      }
+    },
+    {
+      "name": "email_dead_letter",
+      "vhost": "/",
+      "durable": true
+    },
+    {
+      "name": "crawling_full_dead_letter",
+      "vhost": "/",
+      "durable": true
+    }
+  ],
+  "bindings": [
+    {
+      "source": "email",
+      "vhost": "/",
+      "destination": "email_send",
+      "destination_type": "queue",
+      "routing_key": "email.send",
+      "arguments": {}
+    },
+    {
+      "source": "crawling",
+      "vhost": "/",
+      "destination": "crawling_full",
+      "destination_type": "queue",
+      "routing_key": "crawling.full",
+      "arguments": {}
+    },
+    {
+      "source": "dead_letter",
+      "vhost": "/",
+      "destination": "email_dead_letter",
+      "destination_type": "queue",
+      "routing_key": "email.dead_letter",
+      "arguments": {}
+    },
+    {
+      "source": "dead_letter",
+      "vhost": "/",
+      "destination": "crawling_full_dead_letter",
+      "destination_type": "queue",
+      "routing_key": "crawling.full.dead_letter",
+      "arguments": {}
+    }
+  ]
+}

docker-compose/docker-compose.infra.yml

@@ -76,6 +76,8 @@ services:
       RABBITMQ_DEFAULT_PASS: "denamu-rabbitmq-password"
     volumes:
       - denamu-rabbitmq:/var/lib/rabbitmq
+      - ./definitions.json:/etc/rabbitmq/definitions.json:ro
+      - ./rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro
     healthcheck:
       test: [ "CMD", "rabbitmq-diagnostics", "ping" ]
       interval: 30s

docker-compose/docker-compose.local.yml

@@ -36,6 +36,8 @@ services:
         condition: service_healthy
       redis:
         condition: service_healthy
+      rabbitmq:
+        condition: service_healthy
     volumes:
       - ../server/logs:/var/web05-Denamu/server/logs
     environment:
@@ -55,6 +57,8 @@ services:
         condition: service_healthy
       redis:
         condition: service_healthy
+      rabbitmq:
+        condition: service_healthy
     volumes:
       - ../feed-crawler/logs:/var/web05-Denamu/feed-crawler/logs
     environment:

docker-compose/docker-compose.prod.infra.yml

@@ -87,6 +87,8 @@ services:
       - /var/prod_config/infra/.env.prod
     volumes:
       - denamu-rabbitmq:/var/lib/rabbitmq
+      - ./definitions.json:/etc/rabbitmq/definitions.json:ro
+      - ./rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro
     healthcheck:
       test: [ "CMD", "rabbitmq-diagnostics", "ping" ]
       interval: 30s

docker-compose/rabbitmq.conf

@@ -0,0 +1,2 @@
+# Definitions 파일 로드
+management.load_definitions = /etc/rabbitmq/definitions.json

feed-crawler/src/common/rabbitmq.constant.ts

@@ -0,0 +1,25 @@
+export const RMQ_EXCHANGES = {
+  EMAIL: 'EmailExchange',
+  CRAWLING: 'CrawlingExchange',
+  DEAD_LETTER: 'DeadLetterExchange',
+};
+
+export const RMQ_QUEUES = {
+  EMAIL_SEND: 'email.send.queue',
+  CRAWLING_FULL: 'crawling.full.queue',
+  EMAIL_DEAD_LETTER: 'email.deadLetter.queue',
+  CRAWLING_FULL_DEAD_LETTER: 'crawling.full.deadLetter.queue',
+};
+
+export const RMQ_ROUTING_KEYS = {
+  EMAIL_SEND: 'email.send',
+  CRAWLING_FULL: 'crawling.full',
+  EMAIL_DEAD_LETTER: 'email.deadLetter',
+  CRAWLING_FULL_DEAD_LETTER: 'crawling.full.deadLetter',
+};
+
+export const RMQ_EXCHANGE_TYPE = {
+  DIRECT: 'direct',
+  TOPIC: 'topic',
+  FANOUT: 'fanout',
+};

feed-crawler/src/common/rabbitmq.manager.ts

@@ -0,0 +1,58 @@
+import { injectable } from 'tsyringe';
+import { Channel, ChannelModel } from 'amqplib';
+import * as amqp from 'amqplib';
+
+@injectable()
+export class RabbitMQManager {
+  private connection: ChannelModel | null;
+  private channel: Channel | null;
+  private connectionPromise: Promise<ChannelModel> | null = null;
+  private channelPromise: Promise<Channel> | null = null;
+
+  constructor() {
+    this.connection = null;
+    this.channel = null;
+  }
+
+  async connect() {
+    if (this.connection) return this.connection;
+    if (this.connectionPromise) return this.connectionPromise;
+
+    this.connectionPromise = amqp.connect({
+      protocol: 'amqp',
+      hostname: process.env.RABBITMQ_HOST,
+      port: Number.parseInt(process.env.RABBITMQ_PORT),
+      username: process.env.RABBITMQ_DEFAULT_USER,
+      password: process.env.RABBITMQ_DEFAULT_PASS,
+    });
+
+    this.connection = await this.connectionPromise;
+    this.connectionPromise = null;
+    return this.connection;
+  }
+
+  async getChannel() {
+    if (this.channel) return this.channel;
+    if (this.channelPromise) return this.channelPromise;
+
+    if (!this.connection) {
+      await this.connect();
+    }
+    this.channelPromise = this.connection.createChannel();
+    this.channel = await this.channelPromise;
+    this.channelPromise = null;
+    return this.channel;
+  }
+
+  async disconnect() {
+    if (this.channel) {
+      await this.channel.close();
+      this.channel = null;
+    }
+
+    if (this.connection) {
+      await this.connection.close();
+      this.connection = null;
+    }
+  }
+}

feed-crawler/src/common/rmq-access.ts

@@ -1,25 +1,46 @@
-import { injectable } from 'tsyringe';
-import * as amqp from 'amqplib';
-import { Channel, ChannelModel } from 'amqplib';
+import { inject, injectable } from 'tsyringe';
+import { RabbitMQManager } from './rabbitmq.manager';
+import { ConsumeMessage } from 'amqplib/properties';
+import { DEPENDENCY_SYMBOLS } from '../types/dependency-symbols';
+import logger from './logger';
 
 @injectable()
 export class RabbitMQConnection {
-  private connection: ChannelModel;
-  private channel: Channel;
   private nameTag: string;
 
-  constructor() {
+  constructor(
+    @inject(DEPENDENCY_SYMBOLS.RabbitMQManager)
+    private readonly rabbitMQManager: RabbitMQManager,
+  ) {
     this.nameTag = '[RabbitMQ]';
-    this.connect();
   }
 
-  async connect() {
-    this.connection = await amqp.connect({
-      protocol: 'amqp',
-      hostname: process.env.RABBITMQ_HOST,
-      port: Number.parseInt(process.env.RABBITMQ_PORT),
-      username: process.env.RABBITMQ_DEFAULT_USER,
-      password: process.env.RABBITMQ_DEFAULT_PASS,
+  async sendMessage(exchange: string, routingKey: string, message: string) {
+    const channel = await this.rabbitMQManager.getChannel();
+    channel.publish(exchange, routingKey, Buffer.from(message));
+  }
+
+  async consumeMessage(
+    queue: string,
+    onMessage: (msg: ConsumeMessage | null) => void | Promise<void>,
+  ) {
+    const channel = await this.rabbitMQManager.getChannel();
+    await channel.consume(queue, async (message) => {
+      try {
+        if (!message) return;
+
+        const parsedMessage = JSON.parse(message.content.toString());
+        await onMessage(parsedMessage);
+
+        channel.ack(message);
+      } catch (error) {
+        logger.error(
+          `${this.nameTag} 메시지 처리 중 오류 발생
+          오류 메시지: ${error.message}
+          스택 트레이스: ${error.stack}`,
+        );
+        channel.nack(message, false, false);
+      }
     });
   }
 }

feed-crawler/src/container.ts

@@ -14,6 +14,7 @@ import { Atom10Parser } from './common/parser/formats/atom10-parser';
 import { FeedCrawler } from './feed-crawler';
 import { FullFeedCrawlEventWorker } from './event_worker/workers/full-feed-crawl-event-worker';
 import { RabbitMQConnection } from './common/rmq-access';
+import { RabbitMQManager } from './common/rabbitmq.manager';
 
 container.registerSingleton<DatabaseConnection>(
   DEPENDENCY_SYMBOLS.DatabaseConnection,
@@ -80,4 +81,9 @@ container.registerSingleton<RabbitMQConnection>(
   RabbitMQConnection,
 );
 
+container.registerSingleton<RabbitMQManager>(
+  DEPENDENCY_SYMBOLS.RabbitMQManager,
+  RabbitMQManager,
+);
+
 export { container };

feed-crawler/src/main.ts

@@ -9,6 +9,7 @@ import { ClaudeEventWorker } from './event_worker/workers/claude-event-worker';
 import * as schedule from 'node-schedule';
 import { RedisConnection } from './common/redis-access';
 import { FullFeedCrawlEventWorker } from './event_worker/workers/full-feed-crawl-event-worker';
+import { RabbitMQManager } from './common/rabbitmq.manager';
 
 function initializeDependencies() {
   return {
@@ -25,6 +26,9 @@ function initializeDependencies() {
     fullFeedCrawlEventWorker: container.resolve<FullFeedCrawlEventWorker>(
       DEPENDENCY_SYMBOLS.FullFeedCrawlEventWorker,
     ),
+    rabbitMQManager: container.resolve<RabbitMQManager>(
+      DEPENDENCY_SYMBOLS.RabbitMQManager,
+    ),
   };
 }
 
@@ -59,18 +63,38 @@ async function handleShutdown(
   logger.info(`${signal} 신호 수신, feed-crawler 종료 중...`);
   await dependencies.dbConnection.end();
   await dependencies.redisConnection.quit();
-  logger.info('DB 및 Redis 연결 종료');
+  await dependencies.rabbitMQManager.disconnect();
+  logger.info('DB, Redis, RabbitMQ 연결 종료');
   process.exit(0);
 }
 
-function startScheduler() {
+async function startScheduler() {
   logger.info('[Feed Crawler Scheduler Start]');
 
   const dependencies = initializeDependencies();
+  await initializeRabbitMQ(dependencies);
   registerSchedulers(dependencies);
 
   process.on('SIGINT', () => handleShutdown(dependencies, 'SIGINT'));
   process.on('SIGTERM', () => handleShutdown(dependencies, 'SIGTERM'));
 }
 
-startScheduler();
+async function initializeRabbitMQ(
+  dependencies: ReturnType<typeof initializeDependencies>,
+) {
+  try {
+    logger.info(`RabbitMQ 초기화 시작...`);
+
+    await dependencies.rabbitMQManager.connect();
+
+    logger.info(`RabbitMQ 초기화 완료`);
+  } catch (error) {
+    logger.error(`RabbitMQ 초기화 실패:`, error);
+    throw error;
+  }
+}
+
+startScheduler().catch((error) => {
+  logger.error(`스케줄러 시작 실패: `, error);
+  process.exit(1);
+});