Merge pull request #91 from yomankum-project/feature/85-socket

feat: webSocket

build.gradle

@@ -41,6 +41,8 @@ dependencies {
 	implementation 'org.springdoc:springdoc-openapi-starter-webmvc-ui:2.0.2'
 	implementation 'com.github.ulisesbocchio:jasypt-spring-boot-starter:3.0.5'
 
+	implementation 'org.springframework.boot:spring-boot-starter-websocket'
+
 	implementation group: 'org.springframework.boot', name: 'spring-boot-starter-webflux', version: '3.0.2'
 
 	implementation 'io.jsonwebtoken:jjwt-api:0.12.5'

src/main/java/com/account/yomankum/kafka/KafkaService.java

@@ -1,17 +1,52 @@
 package com.account.yomankum.kafka;
 
+import com.account.yomankum.auth.common.Auth;
+import com.account.yomankum.auth.common.LoginUser;
+import com.account.yomankum.common.exception.BadRequestException;
+import com.account.yomankum.common.exception.Exception;
+import com.account.yomankum.kafka.dto.AccountBookCreateNotice;
+import com.account.yomankum.kafka.dto.AccountBookInputNotice;
+import com.account.yomankum.kafka.dto.AccountBookUpdateNotice;
+import com.account.yomankum.socket.common.CustomWebSocketHandler;
+import com.account.yomankum.socket.dto.AccountBookWebSocketNotice;
+import com.account.yomankum.user.domain.User;
+import com.account.yomankum.user.service.UserFinder;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
 @Slf4j
 @Component
+@RequiredArgsConstructor
 public class KafkaService {
 
-    @KafkaListener(topics = "accountBook_input", groupId = "accountServiceConsumers")
-    public void listenAccountNotifications(AccountBookInputNotice notice) {
-        // 처리 로직
-        System.out.println("Received Account Notification: " + notice.nickname());
+    private final UserFinder userFinder;
+    private final CustomWebSocketHandler customWebSocketHandler;
+
+    @KafkaListener(topics = "input", groupId = "accountBook")
+    public void inputAccountBookNotification(AccountBookInputNotice notice) {
+        customWebSocketHandler.sendAccountBookMessage(AccountBookWebSocketNotice.from(notice));
+        log.info("[Kafka] input 메시지 수신 - accountBookId : {}", notice.accountBookId());
+    }
+
+    @KafkaListener(topics = "create", groupId = "accountBook")
+    public void createAccountBookNotification(AccountBookCreateNotice notice) {
+        // 알림 추가 -> Http ?
+        log.info("[Kafka] create 메시지 수신 - accountBookId : {}", notice.accountBookId());
+    }
+
+    @KafkaListener(topics = "update", groupId = "accountBook")
+    public void updateAccountBookNotification(@Auth LoginUser loginUser, AccountBookUpdateNotice notice) {
+        User user = userFinder.findById(loginUser.getUserId())
+                .orElseThrow(() -> new BadRequestException(Exception.USER_NOT_FOUND));
+
+        boolean isExist = user.isUsersAccountBook(notice.accountBookId());
+
+        if (isExist) {
+            // 알림 추가 -> Http ?
+            log.info("[Kafka] update 메시지 수신 - accountBookId : {}", notice.accountBookId());
+        }
     }
 }
 

src/main/java/com/account/yomankum/kafka/dto/AccountBookCreateNotice.java

@@ -0,0 +1,11 @@
+package com.account.yomankum.kafka.dto;
+
+import java.time.LocalDateTime;
+
+public record AccountBookCreateNotice(
+        Long accountBookId,
+        String nickname, // 쓰기 시작하는 닉네임
+        LocalDateTime accountBookCreatedAt,
+        LocalDateTime eventCreatedAt
+) {
+}

src/main/java/com/account/yomankum/kafka/dto/AccountBookInputNotice.java

@@ -0,0 +1,11 @@
+package com.account.yomankum.kafka.dto;
+
+import java.time.LocalDateTime;
+
+public record AccountBookInputNotice(
+        Long accountBookId,
+        String nickname, // 쓰기 시작하는 닉네임
+        LocalDateTime accountBookCreatedAt,
+        LocalDateTime eventCreatedAt
+) {
+}

src/main/java/com/account/yomankum/kafka/dto/AccountBookUpdateNotice.java

@@ -0,0 +1,11 @@
+package com.account.yomankum.kafka.dto;
+
+import java.time.LocalDateTime;
+
+public record AccountBookUpdateNotice(
+        Long accountBookId,
+        String nickname, // 쓰기 시작하는 닉네임
+        LocalDateTime accountBookCreatedAt,
+        LocalDateTime eventCreatedAt
+) {
+}

src/main/java/com/account/yomankum/socket/common/CustomHttpSessionHandshakeInterceptor.java

@@ -0,0 +1,26 @@
+package com.account.yomankum.socket.common;
+
+import jakarta.servlet.http.HttpServletRequest;
+import org.springframework.http.server.ServerHttpRequest;
+import org.springframework.http.server.ServerHttpResponse;
+import org.springframework.util.StringUtils;
+import org.springframework.web.socket.WebSocketHandler;
+import org.springframework.web.socket.server.support.HttpSessionHandshakeInterceptor;
+
+import java.util.Map;
+
+public class CustomHttpSessionHandshakeInterceptor extends HttpSessionHandshakeInterceptor {
+
+    @Override
+    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
+        HttpServletRequest httpReq = (HttpServletRequest) request;
+        String userId = httpReq.getParameter("userId");
+
+        if (StringUtils.hasText(userId)) {
+            attributes.put("userId", userId);
+        }
+
+        return super.beforeHandshake(request, response, wsHandler, attributes);
+    }
+
+}

src/main/java/com/account/yomankum/socket/common/CustomWebSocketHandler.java

@@ -0,0 +1,96 @@
+package com.account.yomankum.socket.common;
+
+import com.account.yomankum.common.exception.BadRequestException;
+import com.account.yomankum.common.exception.Exception;
+import com.account.yomankum.socket.dto.AccountBookWebSocketNotice;
+import com.account.yomankum.user.domain.User;
+import com.account.yomankum.user.service.UserFinder;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+import org.springframework.web.socket.CloseStatus;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.handler.TextWebSocketHandler;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArraySet;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class CustomWebSocketHandler extends TextWebSocketHandler {
+    // 소켓에 연결되어있는 세션 목록
+    private final CopyOnWriteArraySet<WebSocketSession> sessions = new CopyOnWriteArraySet<>();
+    private final Map<Long, WebSocketSession> userIdSessionMap = new ConcurrentHashMap<>();
+
+    private final UserFinder userFinder;
+    private final ObjectMapper mapper;
+
+    @Override
+    public void afterConnectionEstablished(WebSocketSession session) {
+        sessions.add(session);
+        String userId = (String) session.getAttributes().get("userId");
+
+        if (StringUtils.hasText(userId)) {
+            userIdSessionMap.put(Long.parseLong(userId), session);
+        }
+
+        log.info("소켓 연결 성공 - userID : {}", userId);
+    }
+
+    @Override
+    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
+        sessions.remove(session);
+        String userId = (String) session.getAttributes().get("userId");
+
+        if (StringUtils.hasText(userId)) {
+            userIdSessionMap.remove(Long.parseLong(userId), session);
+        }
+
+        log.info("소켓 연결 종료 - userID : {}", userId);
+    }
+
+    @Override
+    protected void handleTextMessage(WebSocketSession session, TextMessage message) {
+        // message 받으면..
+    }
+
+    // 클라이언트에게 메시지 보내기
+    public void sendAccountBookMessage(AccountBookWebSocketNotice notice) {
+        List<Long> userIdList = new ArrayList<>(userIdSessionMap.keySet());
+        List<User> userList = userFinder.findAllById(userIdList);
+
+        Long accountBookId = notice.accountBookId();
+        List<Long> accountBookUserIdList = userList.stream()
+                .filter(user -> user.isUsersAccountBook(accountBookId))
+                .map(User::getId)
+                .toList();
+
+        accountBookUserIdList.forEach(userId -> {
+            WebSocketSession session = userIdSessionMap.get(userId);
+
+            if (session != null && session.isOpen()) {
+                try {
+                    String message = mapper.writeValueAsString(notice);
+                    session.sendMessage(new TextMessage(message));
+                } catch (JsonProcessingException e) {
+                    log.error("JSON 파싱 에러");
+                    throw new BadRequestException(Exception.SERVER_ERROR);
+                } catch (IOException e) {
+                    log.error("세션 메시지 전송 에러");
+                    throw new BadRequestException(Exception.SERVER_ERROR);
+                }
+            }
+        });
+
+    }
+
+}

src/main/java/com/account/yomankum/socket/common/WebSocketConfig.java

@@ -0,0 +1,24 @@
+package com.account.yomankum.socket.common;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.config.annotation.EnableWebSocket;
+import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
+import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
+
+@Configuration
+@EnableWebSocket
+@RequiredArgsConstructor
+public class WebSocketConfig implements WebSocketConfigurer {
+
+    private final CustomWebSocketHandler customWebSocketHandler;
+
+    @Override
+    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
+        registry.addHandler(customWebSocketHandler, "/ws")
+                .setAllowedOrigins("https://www.yomankum.com") // FE Server 오리진 추가..
+                .addInterceptors(new CustomHttpSessionHandshakeInterceptor())
+                .withSockJS();
+    }
+
+}

src/main/java/com/account/yomankum/socket/dto/AccountBookWebSocketNotice.java

@@ -0,0 +1,19 @@
+package com.account.yomankum.socket.dto;
+
+import com.account.yomankum.kafka.dto.AccountBookInputNotice;
+import lombok.Builder;
+
+@Builder
+public record AccountBookWebSocketNotice(
+        Long accountBookId,
+        String nickname,
+        EventType eventType
+) {
+    public static AccountBookWebSocketNotice from(AccountBookInputNotice notice) {
+        return AccountBookWebSocketNotice.builder()
+                .accountBookId(notice.accountBookId())
+                .eventType(EventType.ACCOUNT_BOOK_INPUT)
+                .nickname(notice.nickname())
+                .build();
+    }
+}

src/main/java/com/account/yomankum/socket/dto/EventType.java

@@ -0,0 +1,5 @@
+package com.account.yomankum.socket.dto;
+
+public enum EventType {
+    ACCOUNT_BOOK_INPUT
+}

src/main/java/com/account/yomankum/user/domain/User.java

@@ -1,17 +1,15 @@
 package com.account.yomankum.user.domain;
 
-import com.account.yomankum.user.dto.request.UserInfoUpdateRequest;
 import com.account.yomankum.accountBook.domain.AccountBookUser;
-import com.account.yomankum.common.exception.BadRequestException;
-import com.account.yomankum.common.exception.Exception;
+import com.account.yomankum.user.dto.request.UserInfoUpdateRequest;
 import jakarta.persistence.*;
 import lombok.*;
 import org.springframework.security.crypto.password.PasswordEncoder;
 
 import java.time.LocalDate;
+import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.List;
-import java.time.LocalDateTime;
 
 @Entity
 @Getter
@@ -88,4 +86,10 @@ public String getOauthId() {
     public void addAccountBook(AccountBookUser accountBookUser) {
         accountBooks.add(accountBookUser);
     }
+
+    public boolean isUsersAccountBook(Long accountBookId) {
+        return accountBooks.stream()
+                .anyMatch(accountBook ->
+                        accountBookId.equals(accountBook.getAccountBook().getId()));
+    }
 }

src/main/java/com/account/yomankum/user/service/UserFinder.java

@@ -10,6 +10,7 @@
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.util.List;
 import java.util.Optional;
 
 @Component
@@ -37,4 +38,7 @@ public Optional<User> findByAuthTypeAndOauthId(AuthType type, String oauthId){
         return userRepository.findByAuthInfoAuthTypeAndAuthInfoOauthId(type, oauthId);
     }
 
+    public List<User> findAllById(List<Long> userIdList) {
+        return userRepository.findAllById(userIdList);
+    }
 }

src/main/resources/application.yml

@@ -65,4 +65,4 @@ auth:
 kafka:
   bootstrap-server: localhost:29092
   group-id: accountBook
-  type-mappings: yomankum.kafka.producer.yomankum.api.dto.AccountBookInputNotice:com.account.yomankum.kafka.AccountBookInputNotice
+  type-mappings: yomankum.kafka.producer.yomankum.api.dto.AccountBookCreateNotice:com.account.yomankum.kafka.dto.AccountBookCreateNotice, yomankum.kafka.producer.yomankum.api.dto.AccountBookUpdateNotice:com.account.yomankum.kafka.dto.AccountBookUpdateNotice, yomankum.kafka.producer.yomankum.api.dto.AccountBookInputNotice:com.account.yomankum.kafka.dto.AccountBookInputNotice