feature: Kafka Producer - 결과 메시지 발송 기능

build.gradle

@@ -97,6 +97,10 @@ dependencies {
 	annotationProcessor "com.querydsl:querydsl-apt:${dependencyManagement.importedProperties['querydsl.version']}:jakarta"
 	annotationProcessor 'jakarta.annotation:jakarta.annotation-api'
 	annotationProcessor 'jakarta.persistence:jakarta.persistence-api'
+
+	// kafka
+	implementation 'org.springframework.kafka:spring-kafka:3.1.0'
+
 }
 
 tasks.named('test') {

src/main/java/com/devspacehub/ast/common/constant/ProfileType.java

@@ -24,11 +24,22 @@ public enum ProfileType {
     private final String code;
     private final String accountStatus;
 
-
+    /**
+     * 실행 환경 별 계좌 상태를 반환한다.
+     * @return 계좌 상태
+     */
     public static String getAccountStatus() {
         return PROD.code.equals(getActiveProfile()) ? PROD.accountStatus : TEST.accountStatus;
     }
 
+    /**
+     * 실행 환경 별 코드를 반환한다.
+     * @return 코드
+     */
+    public static ProfileType getActiveProfileType() {
+        return PROD.code.equals(getActiveProfile()) ? PROD : TEST;
+    }
+
     /**
      * 동작 중인 애플리케이션이 'Prod'이면 True를 반환한다. 반대는 False.
      * @return 'Prod' 여부 Boolean 값.
@@ -44,7 +55,7 @@ public static Boolean isProdActive() {
      * 현재 동작 중인 애플리케이션의 Profile을 반환한다.
      * @return 애플리케이션의 profile
      */
-    static String getActiveProfile() {
+    private static String getActiveProfile() {
         return System.getProperty("spring.profiles.active");
     }
 }

src/main/java/com/devspacehub/ast/domain/mashup/service/MashupService.java

@@ -15,7 +15,7 @@
 import com.devspacehub.ast.domain.my.service.MyService;
 import com.devspacehub.ast.domain.my.service.MyServiceFactory;
 import com.devspacehub.ast.domain.notification.Notificator;
-import com.devspacehub.ast.domain.notification.dto.MessageContentDto;
+import com.devspacehub.ast.domain.notification.dto.ItemConclusionResultDto;
 import com.devspacehub.ast.domain.oauth.service.OAuthService;
 import com.devspacehub.ast.domain.orderTrading.OrderTrading;
 import com.devspacehub.ast.domain.orderTrading.OrderTradingServiceFactory;
@@ -78,7 +78,7 @@ public void startOrderConclusionResultProcess(OpenApiType openApiType) {
             myServiceImpl.updateMyReservationOrderUseYn(orderConclusion, today);
 
             // 체결된 종목들에 대해 디스코드 메시지 전달
-            notificator.sendMessage(MessageContentDto.ConclusionResult.fromOne(openApiType, getAccountStatus(), orderConclusion));
+            notificator.sendStockResultMessage(ItemConclusionResultDto.from(openApiType, getAccountStatus(), orderConclusion));
             RequestUtil.timeDelay();
         }
     }

src/main/java/com/devspacehub/ast/domain/my/reservationOrderInfo/service/OverseasReservationBuyOrderService.java

@@ -21,7 +21,7 @@
 import com.devspacehub.ast.domain.my.service.MyService;
 import com.devspacehub.ast.domain.my.service.MyServiceFactory;
 import com.devspacehub.ast.domain.notification.Notificator;
-import com.devspacehub.ast.domain.notification.dto.MessageContentDto;
+import com.devspacehub.ast.domain.notification.dto.DefaultItemInfoDto.ItemOrderResultDto;
 import com.devspacehub.ast.domain.orderTrading.OrderTrading;
 import com.devspacehub.ast.domain.orderTrading.OrderTradingRepository;
 import com.devspacehub.ast.domain.orderTrading.dto.OverseasStockOrderApiReqDto;
@@ -134,7 +134,7 @@ private OrderTrading buy(ReservationStockItem reservationItem, OpenApiProperties
     /**
      * 예약 종목 데이터 validation check
      * @param reservation 예약 종목 데이터
-     * @return 유효하면 True
+     * @throws InvalidValueException 값이 유효하지 않은 경우
      */
     protected void validateValue(ReservationStockItem reservation) {
         if (StringUtils.isBlank(reservation.getItemCode())) {
@@ -197,7 +197,7 @@ private BigDecimal getMyDeposit(ReservationStockItem reservation) {
     /**
      * 예수금과 주문할 총 가격을 비교하여 주문 가능한지 체크한다.
      * @param reservation 해외 예약 종목 정보 Dto
-     * @return 주문 가능한지 여부 boolean 값
+     * @throws InsufficientMoneyException 잔고가 부족한 경우
      */
     protected void sufficientDepositCheck(ReservationStockItem reservation, BigDecimal myDeposit) {
         StockItemDto overseasStockItem = reservation.getStockItem();
@@ -261,7 +261,7 @@ public List<OrderTrading> saveOrderInfos(List<OrderTrading> orderTradingInfos) {
     @Override
     public void orderApiResultProcess(OrderTrading orderTrading) {
         LogUtils.tradingOrderSuccess(OVERSEAS_STOCK_RESERVATION_BUY_ORDER, orderTrading.getItemNameKor());
-        notificator.sendMessage(MessageContentDto.OrderResult.fromOne(OVERSEAS_STOCK_RESERVATION_BUY_ORDER, getAccountStatus(), orderTrading));
+        notificator.sendStockResultMessage(ItemOrderResultDto.from(OVERSEAS_STOCK_RESERVATION_BUY_ORDER, getAccountStatus(), orderTrading));
     }
 
     /**

src/main/java/com/devspacehub/ast/domain/my/reservationOrderInfo/service/ReservationBuyOrderServiceImpl.java

@@ -23,7 +23,7 @@
 import com.devspacehub.ast.domain.my.service.MyService;
 import com.devspacehub.ast.domain.my.service.MyServiceFactory;
 import com.devspacehub.ast.domain.notification.Notificator;
-import com.devspacehub.ast.domain.notification.dto.MessageContentDto;
+import com.devspacehub.ast.domain.notification.dto.DefaultItemInfoDto.ItemOrderResultDto;
 import com.devspacehub.ast.domain.orderTrading.OrderTrading;
 import com.devspacehub.ast.domain.orderTrading.OrderTradingRepository;
 import com.devspacehub.ast.domain.orderTrading.dto.DomesticStockOrderExternalReqDto;
@@ -216,6 +216,6 @@ public List<OrderTrading> saveOrderInfos(List<OrderTrading> orderTradingInfos) {
     @Override
     public void orderApiResultProcess(OrderTrading orderTrading) {
         LogUtils.tradingOrderSuccess(DOMESTIC_STOCK_RESERVATION_BUY_ORDER, orderTrading.getItemNameKor());
-        notificator.sendMessage(MessageContentDto.OrderResult.fromOne(DOMESTIC_STOCK_RESERVATION_BUY_ORDER, getAccountStatus(), orderTrading));
+        notificator.sendStockResultMessage(ItemOrderResultDto.from(DOMESTIC_STOCK_RESERVATION_BUY_ORDER, getAccountStatus(), orderTrading));
     }
 }

src/main/java/com/devspacehub/ast/domain/notification/Notificator.java

@@ -7,80 +7,36 @@
  */
 package com.devspacehub.ast.domain.notification;
 
-import com.devspacehub.ast.common.constant.OpenApiType;
-import com.devspacehub.ast.common.constant.ResultCode;
-import com.devspacehub.ast.domain.notification.dto.DiscordWebhookNotifyRequestDto;
-import com.devspacehub.ast.domain.notification.dto.MessageContentDto;
-import com.devspacehub.ast.exception.error.NotificationException;
-import com.devspacehub.ast.exception.error.InvalidValueException;
+import com.devspacehub.ast.common.constant.ProfileType;
+import com.devspacehub.ast.infra.kafka.config.KafkaTopicType;
+import com.devspacehub.ast.infra.kafka.dto.MessageDto;
+import com.devspacehub.ast.infra.kafka.producer.MessageProducer;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpStatus;
 import org.springframework.stereotype.Service;
-import org.springframework.web.reactive.function.client.WebClient;
-import org.springframework.web.reactive.function.client.WebClientResponseException;
-import java.util.function.Consumer;
-
 
 /**
- * Discord Webhook 이용한 알림 서비스
+ * 알림 클래스
  */
 @Slf4j
 @Service
+@RequiredArgsConstructor
 public class Notificator {
-    @Value("${notify.discord-webhook.url}")
-    private String discordWebhookUrl;
-    private static final String DOMESTIC_ORDER_NOTI_SENDER_NAME = "국내 주문 봇";
-    private static final String DOMESTIC_CONCLUSION_NOTI_SENDER_NAME = "국내 체결 봇";
-
-    private static final String OVERSEAS_ORDER_NOTI_SENDER_NAME = "해외 주문 봇";
-    private static final String OVERSEAS_CONCLUSION_NOTI_SENDER_NAME = "해외 체결 봇";
+    private final MessageProducer messageProducer;
 
     /**
-     * 단일 메시지 발송 요청
-     * @param content 메세지 내용 DTO
+     * 주식 결과 알림 메세지 발송 요청한다.
+     * @param content MessageDto 상속하는 메세지 Dto 클래스
      */
-    public <T extends MessageContentDto> void sendMessage(T content) {
-        String senderName = getSenderName(content.getOpenApiType());
-        String msg = content.createMessage(content);
-        Consumer<HttpHeaders> headers = DiscordWebhookNotifyRequestDto.setHeaders();
-        DiscordWebhookNotifyRequestDto requestBody = DiscordWebhookNotifyRequestDto.builder()
-                .senderName(senderName)
-                .message(msg)
-                .build();
-
-        try {
-            WebClient.builder()
-                    .baseUrl(discordWebhookUrl)
-                    .build()
-                    .post()
-                    .headers(headers)
-                    .bodyValue(requestBody)
-                    .retrieve()
-                    .toBodilessEntity()
-                    .block();
-
-        } catch (Exception ex) {
-            if (ex instanceof WebClientResponseException webClientResponseException && HttpStatus.NO_CONTENT.equals(webClientResponseException.getStatusCode())) {
-                return;
-            }
-            throw new NotificationException(ex.getMessage());
-        }
+    public <T extends MessageDto> void sendStockResultMessage(T content) {
+        messageProducer.produce(this.createTopic(KafkaTopicType.STOCK_RESULT_NOTI_TOPIC), content);
     }
+
     /**
-     * 알림 봇 이름 지정
-     * @param openApiType OpenApiType
-     * @return 알림 봇 이름
+     * 환경 별 토픽을 생성한다.
+     * @return 토픽
      */
-    private String getSenderName(OpenApiType openApiType) {
-        return switch (openApiType) {
-            case DOMESTIC_STOCK_BUY_ORDER, DOMESTIC_STOCK_SELL_ORDER, DOMESTIC_STOCK_RESERVATION_BUY_ORDER -> DOMESTIC_ORDER_NOTI_SENDER_NAME;
-            case DOMESTIC_ORDER_CONCLUSION_FIND -> DOMESTIC_CONCLUSION_NOTI_SENDER_NAME;
-            case OVERSEAS_STOCK_BUY_ORDER, OVERSEAS_STOCK_SELL_ORDER, OVERSEAS_STOCK_RESERVATION_BUY_ORDER -> OVERSEAS_ORDER_NOTI_SENDER_NAME;
-            case OVERSEAS_ORDER_CONCLUSION_FIND -> OVERSEAS_CONCLUSION_NOTI_SENDER_NAME;
-            default -> throw new InvalidValueException(ResultCode.INVALID_OPENAPI_TYPE_ERROR);
-        };
+    private String createTopic(KafkaTopicType topicType) {
+        return topicType.getValue(ProfileType.getActiveProfileType());
     }
-
 }

src/main/java/com/devspacehub/ast/domain/notification/dto/DefaultItemInfoDto.java

@@ -0,0 +1,65 @@
+/*
+ © 2024 devspacehub, Inc. All rights reserved.
+
+ name : DefaultItemInfoDto
+ creation : 2024.9.18
+ author : Yoonji Moon
+ */
+
+package com.devspacehub.ast.domain.notification.dto;
+
+import com.devspacehub.ast.common.constant.OpenApiType;
+import com.devspacehub.ast.domain.orderTrading.OrderTrading;
+import com.devspacehub.ast.infra.kafka.dto.MessageDto;
+import lombok.Getter;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+import java.math.BigDecimal;
+
+/**
+ * 메세지 발송 시 생성되는 주식 관련 정보 Dto
+ */
+@Getter
+@SuperBuilder
+public abstract class DefaultItemInfoDto extends MessageDto {
+    private String accountStatusKor;
+    private String itemNameKor;
+    private String itemCode;
+    private OpenApiType openApiType;
+    private int orderQuantity;
+    private BigDecimal orderPrice;
+    private String orderNumber;
+    private String orderTime;
+
+    /**
+     * 주문 결과 Dto.
+     * DefaultItemInfoDto를 상속한다.
+     */
+    @ToString
+    @SuperBuilder
+    public static class ItemOrderResultDto extends DefaultItemInfoDto {
+
+        /**
+         * 전달받는 인자로 주문 결과 Dto 인스턴스를 반환한다.
+         * @param openApiType OpenApi Type
+         * @param accountStatus 계좌 상태
+         * @param orderTrading 주문 결과 Dto
+         * @return ItemOrderResultDto
+         */
+        public static ItemOrderResultDto from(OpenApiType openApiType, String accountStatus, OrderTrading orderTrading) {
+            return ItemOrderResultDto.builder()
+                    .title("주문 완료")
+                    .accountStatusKor(accountStatus)
+                    .itemNameKor(orderTrading.getItemNameKor())
+                    .itemCode(orderTrading.getItemCode())
+                    .openApiType(openApiType)
+                    .orderQuantity(orderTrading.getOrderQuantity())
+                    .orderPrice(orderTrading.getOrderPrice())
+                    .orderNumber(orderTrading.getOrderNumber())
+                    .orderTime(orderTrading.getOrderTime())
+                    .build();
+        }
+    }
+
+}

src/main/java/com/devspacehub/ast/domain/notification/dto/ItemConclusionResultDto.java

@@ -0,0 +1,51 @@
+/*
+ © 2024 devspacehub, Inc. All rights reserved.
+
+ name : ItemConclusionResultDto
+ creation : 2024.9.18
+ author : Yoonji Moon
+ */
+
+package com.devspacehub.ast.domain.notification.dto;
+
+import com.devspacehub.ast.common.constant.OpenApiType;
+import com.devspacehub.ast.domain.my.dto.orderConclusion.OrderConclusionDto;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+import java.math.BigDecimal;
+
+/**
+ * 주식 체결 결과 Dto
+ * DefaultItemInfoDto 상속
+ */
+@ToString
+@SuperBuilder
+public class ItemConclusionResultDto extends DefaultItemInfoDto {
+    private int concludedQuantity;
+    private BigDecimal concludedPrice;
+
+
+    /**
+     * 전달받는 인자로 주식 체결 결과 Dto 인스턴스 생성하여 반환한다.
+     * @param orderApiType OpenApi Type
+     * @param accountStatus 계좌 상태
+     * @param orderConclusion 주문 결과 정보
+     * @return ItemConclusionResultDto
+     */
+    public static ItemConclusionResultDto from(OpenApiType orderApiType, String accountStatus, OrderConclusionDto orderConclusion) {
+        return ItemConclusionResultDto.builder()
+                .title("체결 완료")
+                .accountStatusKor(accountStatus)
+                .itemNameKor(orderConclusion.getItemNameKor())
+                .itemCode(orderConclusion.getItemCode())
+                .openApiType(orderApiType)
+                .orderQuantity(orderConclusion.getOrderQuantity())
+                .orderPrice(orderConclusion.getOrderPrice())
+                .orderNumber(orderConclusion.getOrderNumber())
+                .concludedQuantity(orderConclusion.getConcludedQuantity())
+                .concludedPrice(orderConclusion.getConcludedPrice())
+                .orderTime(orderConclusion.getOrderTime())
+                .build();
+    }
+}