채팅서버 카프카 연결

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/config/WebSocketConfig.java

@@ -12,9 +12,7 @@ public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
 
     @Override
     public void configureMessageBroker(MessageBrokerRegistry config) {
-        //스프링 인메모리 메시지 브로커 사용
-        config.enableSimpleBroker("/subscribe");    //구독 prefix
-        config.setApplicationDestinationPrefixes("/publish");   //발행 prefix
+        config.setApplicationDestinationPrefixes("/publish"); // STOMP 메시지 발행 prefix
     }
 
     @Override

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/controller/ChatController.java

@@ -1,24 +1,26 @@
 package com.jootalkpia.chat_server.controller;
 
-import com.jootalkpia.chat_server.dto.ChatMessageRequest;
-import com.jootalkpia.chat_server.dto.ChatMessageResponse;
+import com.jootalkpia.chat_server.dto.messgaeDto.ChatMessageRequest;
+import com.jootalkpia.chat_server.dto.ChatMessageToKafka;
 import com.jootalkpia.chat_server.service.ChatService;
+import com.jootalkpia.chat_server.service.KafkaProducer;
+import java.util.List;
 import lombok.RequiredArgsConstructor;
 import org.springframework.messaging.handler.annotation.DestinationVariable;
 import org.springframework.messaging.handler.annotation.MessageMapping;
-import org.springframework.messaging.handler.annotation.SendTo;
 import org.springframework.web.bind.annotation.RestController;
 
 @RestController
 @RequiredArgsConstructor
 public class ChatController {
 
+    private final KafkaProducer kafkaProducer;
     private final ChatService chatService;
 
     @MessageMapping("/chat.{channelId}")
-    @SendTo("/subscribe/chat.{channelId}")
-    public ChatMessageResponse sendMessage(ChatMessageRequest request, @DestinationVariable Long channelId) {
-        return chatService.createMessage(request.userId(), request.content());
+    public void sendMessage(ChatMessageRequest request, @DestinationVariable Long channelId) {
+        List chatMessage = chatService.createMessage(request);  //Type 건들지말것
+        ChatMessageToKafka chatMessageToKafka = new ChatMessageToKafka(channelId, chatMessage);
+        kafkaProducer.sendChatMessage(chatMessageToKafka, channelId);
     }
-
 }

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/domain/Files.java

@@ -0,0 +1,26 @@
+package com.jootalkpia.chat_server.domain;
+
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import java.time.LocalDateTime;
+import lombok.Getter;
+import lombok.Setter;
+import jakarta.persistence.Id;
+
+@Entity
+@Getter
+@Setter
+public class Files {
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long fileId;
+
+    private String url;
+    private String urlThumbnail;
+    private String fileType;
+    private Long fileSize;
+    private String mimeType;
+    private LocalDateTime createdAt;
+    private LocalDateTime updatedAt;
+}

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/dto/ChatMessageToKafka.java

@@ -1,8 +1,9 @@
 package com.jootalkpia.chat_server.dto;
 
+import java.util.List;
+
 public record ChatMessageToKafka(
-        Long userId,
-        String username,
-        String content
+        Long channelId,
+        List chatMessage    //type 건들지말것
 ) {
 }

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/dto/messgaeDto/ChatMessageRequest.java

@@ -0,0 +1,10 @@
+package com.jootalkpia.chat_server.dto.messgaeDto;
+
+import java.util.List;
+
+public record ChatMessageRequest(
+        Long userId,
+        String content,
+        List<Long> attachmentList
+) {
+}

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/dto/messgaeDto/CommonResponse.java

@@ -0,0 +1,12 @@
+package com.jootalkpia.chat_server.dto.messgaeDto;
+
+public record CommonResponse(
+        Long userId,
+        String userNickname,
+        String userProfileImage
+) implements MessageResponse {
+    @Override
+    public String type() {
+        return "COMMON";
+    }
+}

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/dto/messgaeDto/ImageResponse.java

@@ -0,0 +1,10 @@
+package com.jootalkpia.chat_server.dto.messgaeDto;
+
+public record ImageResponse(
+        String url
+) implements MessageResponse {
+    @Override
+    public String type() {
+        return "IMAGE";
+    }
+}

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/dto/messgaeDto/MessageResponse.java

@@ -0,0 +1,19 @@
+package com.jootalkpia.chat_server.dto.messgaeDto;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+
+@JsonTypeInfo(
+        use = JsonTypeInfo.Id.NAME,
+        include = JsonTypeInfo.As.PROPERTY,
+        property = "type"
+)
+@JsonSubTypes({
+        @JsonSubTypes.Type(value = CommonResponse.class, name = "COMMON"),
+        @JsonSubTypes.Type(value = TextResponse.class, name = "TEXT"),
+        @JsonSubTypes.Type(value = VideoResponse.class, name = "VIDEO"),
+        @JsonSubTypes.Type(value = ImageResponse.class, name = "IMAGE")
+})
+public interface MessageResponse {
+    String type();
+}

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/dto/messgaeDto/TextResponse.java

@@ -0,0 +1,10 @@
+package com.jootalkpia.chat_server.dto.messgaeDto;
+
+public record TextResponse(
+        String content
+) implements MessageResponse {
+    @Override
+    public String type() {
+        return "TEXT";
+    }
+}

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/dto/messgaeDto/VideoResponse.java

@@ -0,0 +1,11 @@
+package com.jootalkpia.chat_server.dto.messgaeDto;
+
+public record VideoResponse(
+        String urlThumbnail,
+        String url
+) implements MessageResponse {
+    @Override
+    public String type() {
+        return "VIDEO";
+    }
+}

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/repository/FileRepository.java

@@ -0,0 +1,10 @@
+package com.jootalkpia.chat_server.repository;
+
+import com.jootalkpia.chat_server.domain.Files;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface FileRepository extends JpaRepository<Files, Long> {
+
+}

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/service/ChatService.java

@@ -1,8 +1,17 @@
 package com.jootalkpia.chat_server.service;
 
+import com.jootalkpia.chat_server.domain.Files;
 import com.jootalkpia.chat_server.domain.User;
-import com.jootalkpia.chat_server.dto.ChatMessageResponse;
+import com.jootalkpia.chat_server.dto.messgaeDto.ChatMessageRequest;
+import com.jootalkpia.chat_server.dto.messgaeDto.CommonResponse;
+import com.jootalkpia.chat_server.dto.messgaeDto.ImageResponse;
+import com.jootalkpia.chat_server.dto.messgaeDto.MessageResponse;
+import com.jootalkpia.chat_server.dto.messgaeDto.TextResponse;
+import com.jootalkpia.chat_server.dto.messgaeDto.VideoResponse;
+import com.jootalkpia.chat_server.repository.FileRepository;
 import com.jootalkpia.chat_server.repository.UserRepository;
+import java.util.ArrayList;
+import java.util.List;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
 
@@ -11,9 +20,32 @@
 public class ChatService {
 
     private final UserRepository userRepository;
+    private final FileRepository fileRepository;
 
-    public ChatMessageResponse createMessage(Long userId, String content){
-        User user = userRepository.findByUserId(userId);
-        return new ChatMessageResponse(user.getNickname(),content);
+    public List<MessageResponse> createMessage(ChatMessageRequest request) {
+        List<MessageResponse> response = new ArrayList<>();
+
+        User user = userRepository.findByUserId(request.userId());
+        response.add(new CommonResponse(user.getUserId(), user.getNickname(), user.getProfileImage()));
+
+        String content = request.content();
+        if (content != null && !content.isEmpty()) {
+            response.add(new TextResponse(content));
+        }
+
+        List<Long> attachmentList = request.attachmentList();
+        if (attachmentList != null && !attachmentList.isEmpty()) {
+            for (Long fileId : attachmentList) {
+                Files file = fileRepository.findById(fileId)
+                        .orElseThrow(() -> new IllegalArgumentException("File not found for fileId: " + fileId)); // todo : 예외 처리 추가
+                switch (file.getFileType()) {
+                    case "IMAGE" -> response.add(new ImageResponse(file.getUrl()));
+                    case "VIDEO" -> response.add(new VideoResponse(file.getUrlThumbnail(),file.getUrl()));
+                    default -> throw new IllegalArgumentException("Unsupported file type: " + file.getFileType()); // todo : 예외 처리 추가
+                }
+            }
+        }
+
+        return response;
     }
 }

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/service/KafkaConsumer.java

@@ -1,13 +1,13 @@
 package com.jootalkpia.chat_server.service;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.jootalkpia.chat_server.dto.ChatMessageResponse;
 import com.jootalkpia.chat_server.dto.ChatMessageToKafka;
 import com.jootalkpia.chat_server.dto.MinutePriceResponse;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
-import org.springframework.messaging.simp.SimpMessageSendingOperations;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.simp.SimpMessagingTemplate;
 import org.springframework.stereotype.Service;
 
@@ -16,22 +16,20 @@
 @RequiredArgsConstructor
 public class KafkaConsumer {
 
-    private final ChatService chatService;
     private final ObjectMapper objectMapper;
-    private final SimpMessagingTemplate messagingTemplate; // SimpMessagingTemplate 주입
-    private final SimpMessageSendingOperations messagingTemplateBroker; // 내부 메시지 브로커 사용
+    private final SimpMessagingTemplate messagingTemplate;
 
     @KafkaListener(
             topics = "${topic.minute}",
             groupId = "${group.minute}"
     )
     public void processMinutePrice(String kafkaMessage) {
-        log.info("Received Kafka message ===> " + kafkaMessage);
+        log.info("Received Kafka minute message ===> {}", kafkaMessage);
         try {
             MinutePriceResponse stockUpdate = objectMapper.readValue(kafkaMessage, MinutePriceResponse.class);
             String stockDataJson = objectMapper.writeValueAsString(stockUpdate);
 
-            messagingTemplateBroker.convertAndSend("/subscribe/stock", stockDataJson);
+            messagingTemplate.convertAndSend("/subscribe/stock", stockDataJson);
 
             log.info("Broadcasted stock data via WebSocket: " + stockDataJson);
 
@@ -41,23 +39,22 @@ public void processMinutePrice(String kafkaMessage) {
     }
 
     @KafkaListener(
-            topics = "${topic.chat}",
+            topics = "jootalkpia.chat.prd.message",
             groupId = "${group.chat}", //추후 그룹 ID에 동적인 컨테이너 ID 삽입
             concurrency = "2"
     )
-    public void processChatMessage(String kafkaMessage) {
-        log.info("Received Kafka message ===> " + kafkaMessage);
-
-        ObjectMapper mapper = new ObjectMapper();
-
+    public void processChatMessage(@Header(KafkaHeaders.RECEIVED_KEY) String channelId, String kafkaMessage) {
+        log.info("Received Kafka message ===> channelId: {}, message: {}", channelId, kafkaMessage);
         try {
-            ChatMessageToKafka chatMessageToKafka = mapper.readValue(kafkaMessage, ChatMessageToKafka.class);
+            ChatMessageToKafka chatMessage = objectMapper.readValue(kafkaMessage, ChatMessageToKafka.class);
+            String chatDataJson = objectMapper.writeValueAsString(chatMessage);
 
-            //로컬 메모리와 유저 ID를 비교하는 로직, 있으면 웹소켓을 통한 데이터 전달 없으면 일단 버림
+            // todo : 로컬 메모리와 유저 ID를 비교하는 로직 추가 필요
+            messagingTemplate.convertAndSend("/subscribe/chat." + channelId, chatDataJson);
+            log.info("Broadcasted chat message via WebSocket: {}", chatDataJson);
 
-            log.info("dto ===> " + chatMessageToKafka.toString());
         } catch (Exception ex) {
-            log.error(ex.getMessage(), ex);
+            log.error("Error processing chat message: {}", ex.getMessage(), ex);
         }
     }
 }

src/backend/chat_server/src/main/java/com/jootalkpia/chat_server/service/KafkaProducer.java

@@ -4,6 +4,7 @@
 import com.jootalkpia.chat_server.dto.ChatMessageToKafka;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Service;
 
@@ -15,14 +16,19 @@ public class KafkaProducer {
     private final Gson gson = new Gson();
     private final KafkaTemplate<String, String> kafkaTemplate;
 
-    public void sendChatMessage(ChatMessageToKafka chatMessageToKafka, Long roomId) {
+    @Value("${topic.chat}")
+    private String chatTopic;
+
+    public void sendChatMessage(ChatMessageToKafka chatMessageToKafka, Long channelId) {
         String jsonChatMessage = gson.toJson(chatMessageToKafka);
-        kafkaTemplate.send("jootalkpia.chat.prd.message", String.valueOf(roomId), jsonChatMessage).whenComplete((result, ex) -> { //키 값 설정으로 순서 보장, 실시간성이 떨어짐, 고민해봐야 할 부분
-            if (ex == null) {
-                log.info(result.toString());
-            } else {
-                log.error(ex.getMessage(), ex); //추후 예외처리
-            }
-        });
+
+        kafkaTemplate.send(chatTopic, String.valueOf(channelId), jsonChatMessage)
+                .whenComplete((result, ex) -> {//키 값 설정으로 순서 보장, 실시간성이 떨어짐, 고민해봐야 할 부분
+                    if (ex == null) {
+                        log.info("Kafka message sent: {}", result.toString());
+                    } else {
+                        log.error("Error sending Kafka message: {}", ex.getMessage(), ex); // todo : 추후 예외처리
+                    }
+                });
     }
-}
+}

src/backend/chat_server/src/main/resources/application.yml

@@ -25,6 +25,7 @@ spring:
 topic:
   minute: jootalkpia.stock.prd.minute
   chat: jootalkpia.chat.prd.message
+
 group:
   minute: minute-price-save-consumer-group
   chat: chat-message-handle-consumer-group