fix: SSE emitter 관리 로직 개선 및 구독 안정성 향상 (#275)

soonge2 · web-flow · commit 4c534fd3caca · 2025-07-22T16:41:28.000+09:00
* refactor: SSE 이벤트에서 수동 JSON 직렬화 제거 (#257) * refactor: emitter 정리 로직 개선 (중복 호출 제거, 메서드 분리, key 파싱 안정화) (#257) - findAllEmitters() 중복 호출 제거로 오버헤드 감소 - emitter 제거 조건 메서드로 분리해 가독성 및 재사용성 향상 - emitterId 파싱 시 split → lastIndexOf로 변경하여 userId 내 '_' 포함 대응 * test: isExpiredEmitterEntry 단위 테스트 추가 (#257) - 오래된 emitter 제거 여부 확인 - 최신 emitter는 유지되는지 확인 - 잘못된 키 형식 및 파싱 실패 케이스 처리 * refactor: EmitterRepository 인터페이스 분리 및 emitter 정리 통합 테스트 추가 (#257) * feat: 로그아웃 시 해당 사용자의 SSE emitter 제거 기능 추가 (#257) * fix: SSE 구독 엔드포인트에 produces 헤더 명시 (#257)
diff --git a/src/main/java/com/moa/moa_server/domain/auth/service/AuthService.java b/src/main/java/com/moa/moa_server/domain/auth/service/AuthService.java
@@ -7,6 +7,7 @@
 import com.moa.moa_server.domain.auth.handler.AuthErrorCode;
 import com.moa.moa_server.domain.auth.handler.AuthException;
 import com.moa.moa_server.domain.auth.service.strategy.OAuthLoginStrategy;
+import com.moa.moa_server.domain.notification.repository.NotificationEmitterRepository;
 import com.moa.moa_server.domain.user.entity.User;
 import java.util.Map;
 import java.util.Set;
@@ -31,6 +32,7 @@ public class AuthService {
   private final Map<String, OAuthLoginStrategy> strategies;
   private final JwtTokenService jwtTokenService;
   private final RefreshTokenService refreshTokenService;
+  private final NotificationEmitterRepository notificationEmitterRepository;
 
   public LoginResult login(String provider, String code, String redirectUri) {
     if (!OAuth.ProviderCode.isSupported(provider)) {
@@ -69,6 +71,8 @@ public TokenRefreshResponse refreshAccessToken(String refreshToken) {
   }
 
   public boolean logout(Long userId) {
+    notificationEmitterRepository.deleteAllByUserId(userId);
+    log.info("[AuthService#logout] userId={} 로그아웃 - SSE emitters 제거 완료", userId);
     return refreshTokenService.deleteRefreshTokenByUserId(
         userId); // true면 SUCCESS, false면 ALREADY_LOGGED_OUT
   }
diff --git a/src/main/java/com/moa/moa_server/domain/global/util/JsonUtil.java b/src/main/java/com/moa/moa_server/domain/global/util/JsonUtil.java
diff --git a/src/main/java/com/moa/moa_server/domain/notification/application/sse/NotificationSseBroadcaster.java b/src/main/java/com/moa/moa_server/domain/notification/application/sse/NotificationSseBroadcaster.java
@@ -1,7 +1,5 @@
 package com.moa.moa_server.domain.notification.application.sse;
 
-import static com.moa.moa_server.domain.global.util.JsonUtil.toJson;
-
 import com.moa.moa_server.domain.notification.dto.NotificationItem;
 import com.moa.moa_server.domain.notification.entity.Notification;
 import com.moa.moa_server.domain.notification.repository.NotificationEmitterRepository;
@@ -18,12 +16,11 @@ public class NotificationSseBroadcaster {
   private final NotificationEmitterRepository emitterRepository;
 
   public void send(Notification notification) {
-    String payload = toJson(NotificationItem.from(notification));
     SseEmitter.SseEventBuilder event =
         SseEmitter.event()
             .id(String.valueOf(notification.getId()))
             .name("notification")
-            .data(payload);
+            .data(NotificationItem.from(notification));
 
     emitterRepository
         .findAllByUserId(notification.getUser().getId())
diff --git a/src/main/java/com/moa/moa_server/domain/notification/application/sse/NotificationSseCleaner.java b/src/main/java/com/moa/moa_server/domain/notification/application/sse/NotificationSseCleaner.java
@@ -1,48 +1,49 @@
 package com.moa.moa_server.domain.notification.application.sse;
 
 import com.moa.moa_server.domain.notification.config.SseProperties;
-import com.moa.moa_server.domain.notification.repository.NotificationEmitterRepository;
+import com.moa.moa_server.domain.notification.repository.EmitterRepository;
+import java.util.Map;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 /** 주기적으로 오래된 SSE Emitter를 정리하는 컴포넌트. */
 @Slf4j
 @Component
 @RequiredArgsConstructor
 public class NotificationSseCleaner {
 
-  private final NotificationEmitterRepository emitterRepository;
+  private final EmitterRepository emitterRepository;
   private final SseProperties sseProperties;
 
   @Scheduled(fixedRateString = "#{@sseProperties.staleCleanInterval}")
   public void cleanUpStaleEmitters() {
     long now = System.currentTimeMillis();
     long threshold = sseProperties.getStaleThreshold();
 
-    int before = emitterRepository.findAllEmitters().size();
-
-    emitterRepository
-        .findAllEmitters()
-        .entrySet()
-        .removeIf(
-            entry -> {
-              String key = entry.getKey(); // user_timestamp
-              String[] parts = key.split("_");
-              if (parts.length != 2) return true;
-
-              try {
-                long createdAt = Long.parseLong(parts[1]);
-                return (now - createdAt) > threshold;
-              } catch (NumberFormatException e) {
-                return true; // 잘못된 형식의 키도 정리
-              }
-            });
+    Map<String, SseEmitter> allEmitters = emitterRepository.findAllEmitters();
+    int before = allEmitters.size();
+
+    allEmitters.entrySet().removeIf(entry -> isExpiredEmitterEntry(entry, now, threshold));
 
     int after = emitterRepository.findAllEmitters().size();
     if (before != after) {
       log.info("[NotificationSseCleaner#cleanUpStaleEmitters] 만료된 emitter 정리: {}개", before - after);
     }
   }
+
+  boolean isExpiredEmitterEntry(Map.Entry<String, SseEmitter> entry, long now, long threshold) {
+    String key = entry.getKey(); // userId_timestamp
+    int idx = key.lastIndexOf('_');
+    if (idx == -1 || idx == key.length() - 1) return true;
+
+    try {
+      long createdAt = Long.parseLong(key.substring(idx + 1));
+      return (now - createdAt) > threshold;
+    } catch (NumberFormatException e) {
+      return true; // 잘못된 형식의 키도 정리
+    }
+  }
 }
diff --git a/src/main/java/com/moa/moa_server/domain/notification/application/sse/NotificationSseConnectionHelper.java b/src/main/java/com/moa/moa_server/domain/notification/application/sse/NotificationSseConnectionHelper.java
@@ -1,7 +1,5 @@
 package com.moa.moa_server.domain.notification.application.sse;
 
-import static com.moa.moa_server.domain.global.util.JsonUtil.toJson;
-
 import com.moa.moa_server.domain.notification.dto.NotificationItem;
 import com.moa.moa_server.domain.notification.entity.Notification;
 import com.moa.moa_server.domain.notification.repository.NotificationEmitterRepository;
@@ -39,12 +37,11 @@ public void sendLostEvents(Long userId, String lastEventId, SseEmitter emitter)
 
     for (Notification notification : lostEvents) {
       try {
-        String payload = toJson(NotificationItem.from(notification));
         SseEmitter.SseEventBuilder event =
             SseEmitter.event()
                 .id(String.valueOf(notification.getId()))
                 .name("notification")
-                .data(payload);
+                .data(NotificationItem.from(notification));
         emitter.send(event);
       } catch (IOException e) {
         log.warn("초기 SSE 이벤트 전송 실패. emitter는 유지", e); // 초기 전송 실패는 일시적일 수 있으므로 무시하고 연결 유지
diff --git a/src/main/java/com/moa/moa_server/domain/notification/controller/NotificationController.java b/src/main/java/com/moa/moa_server/domain/notification/controller/NotificationController.java
@@ -11,6 +11,7 @@
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
 import lombok.RequiredArgsConstructor;
+import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.security.core.annotation.AuthenticationPrincipal;
 import org.springframework.web.bind.annotation.*;
@@ -45,7 +46,7 @@ public ResponseEntity<ApiResponse<NotificationReadResponse>> readNotification(
   }
 
   @Operation(summary = "알림 SSE", description = "알림 SSE 커넥션을 열고, SseEmitter를 반환합니다.")
-  @GetMapping("/subscribe")
+  @GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
   public SseEmitter subscribe(
       @AuthenticationPrincipal Long userId,
       @RequestHeader(value = "Last-Event-ID", required = false) String lastEventId,
diff --git a/src/main/java/com/moa/moa_server/domain/notification/repository/EmitterRepository.java b/src/main/java/com/moa/moa_server/domain/notification/repository/EmitterRepository.java
@@ -0,0 +1,14 @@
+package com.moa.moa_server.domain.notification.repository;
+
+import java.util.Map;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+public interface EmitterRepository {
+  void save(String id, SseEmitter emitter);
+
+  void deleteById(String id);
+
+  Map<String, SseEmitter> findAllEmitters();
+
+  public Map<String, SseEmitter> findAllByUserId(Long userId);
+}
diff --git a/src/main/java/com/moa/moa_server/domain/notification/repository/NotificationEmitterRepository.java b/src/main/java/com/moa/moa_server/domain/notification/repository/NotificationEmitterRepository.java
@@ -12,7 +12,7 @@
 
 @Component
 @RequiredArgsConstructor
-public class NotificationEmitterRepository {
+public class NotificationEmitterRepository implements EmitterRepository {
 
   private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
 
@@ -36,6 +36,10 @@ public Map<String, SseEmitter> findAllByUserId(Long userId) {
         .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
   }
 
+  public void deleteAllByUserId(Long userId) {
+    findAllByUserId(userId).keySet().forEach(this::deleteById);
+  }
+
   @Transactional(readOnly = true)
   public List<Notification> findCachedEventsAfter(Long userId, String lastEventId) {
     Long lastId = Long.parseLong(lastEventId);
diff --git a/src/test/java/com/moa/moa_server/domain/notification/application/sse/NotificationSseCleanerIntegrationTest.java b/src/test/java/com/moa/moa_server/domain/notification/application/sse/NotificationSseCleanerIntegrationTest.java
@@ -0,0 +1,48 @@
+package com.moa.moa_server.domain.notification.application.sse;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.moa.moa_server.domain.notification.config.SseProperties;
+import com.moa.moa_server.domain.notification.repository.FakeEmitterRepository;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+public class NotificationSseCleanerIntegrationTest {
+
+  FakeEmitterRepository fakeRepo;
+  SseProperties props;
+  NotificationSseCleaner cleaner;
+
+  @BeforeEach
+  void setUp() {
+    fakeRepo = new FakeEmitterRepository();
+
+    props = new SseProperties();
+    props.setStaleThreshold(1000L); // 1초 넘으면 stale
+    props.setStaleCleanInterval(5000L); // 무관
+
+    cleaner = new NotificationSseCleaner(fakeRepo, props);
+  }
+
+  @Test
+  @DisplayName("오래된 emitter는 cleanUpStaleEmitters 실행 시 제거됨")
+  void testOldEmitterIsRemoved() {
+    long now = System.currentTimeMillis();
+
+    String staleKey = "user1_" + (now - 5000);
+    String validKey = "user2_" + now;
+
+    fakeRepo.save(staleKey, new SseEmitter());
+    fakeRepo.save(validKey, new SseEmitter());
+
+    // when
+    cleaner.cleanUpStaleEmitters();
+
+    // then
+    assertThat(fakeRepo.contains(staleKey)).isFalse();
+    assertThat(fakeRepo.contains(validKey)).isTrue();
+    assertThat(fakeRepo.findAllEmitters()).hasSize(1);
+  }
+}
diff --git a/src/test/java/com/moa/moa_server/domain/notification/application/sse/NotificationSseCleanerTest.java b/src/test/java/com/moa/moa_server/domain/notification/application/sse/NotificationSseCleanerTest.java
@@ -0,0 +1,64 @@
+package com.moa.moa_server.domain.notification.application.sse;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.AbstractMap;
+import java.util.Map;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+/** NotificationSseCleaner의 emitter 제거 조건 판단 메서드(isExpiredEmitterEntry) 단위 테스트. */
+public class NotificationSseCleanerTest {
+
+  NotificationSseCleaner cleaner;
+
+  @BeforeEach
+  void setUp() {
+    cleaner = new NotificationSseCleaner(null, null); // emitterRepository, sseProperties 필요 없음
+  }
+
+  @Test
+  @DisplayName("오래된 emitter는 제거 (true 반환)")
+  void testExpiredEmitterEntry_shouldReturnTrue_ifTimestampTooOld() {
+    long now = System.currentTimeMillis();
+    long threshold = 1000L;
+
+    String key = "123_" + (now - 2000); // 오래된 timestamp
+    Map.Entry<String, SseEmitter> entry = new AbstractMap.SimpleEntry<>(key, new SseEmitter());
+
+    boolean result = cleaner.isExpiredEmitterEntry(entry, now, threshold);
+    assertThat(result).isTrue();
+  }
+
+  @Test
+  @DisplayName("최신 emitter는 제거 대상 아님 (false 반환)")
+  void testExpiredEmitterEntry_shouldReturnFalse_ifTimestampRecent() {
+    long now = System.currentTimeMillis();
+    long threshold = 5000L;
+
+    String key = "123_" + now; // 방금 생성됨
+    Map.Entry<String, SseEmitter> entry = new AbstractMap.SimpleEntry<>(key, new SseEmitter());
+
+    boolean result = cleaner.isExpiredEmitterEntry(entry, now, threshold);
+    assertThat(result).isFalse();
+  }
+
+  @Test
+  @DisplayName("잘못된 키 형식 또는 파싱 실패 시 제거 대상 (true 반환)")
+  void testExpiredEmitterEntry_shouldReturnTrue_ifKeyMalformed() {
+    Map.Entry<String, SseEmitter> entry1 =
+        new AbstractMap.SimpleEntry<>("invalidKey", new SseEmitter());
+    Map.Entry<String, SseEmitter> entry2 = new AbstractMap.SimpleEntry<>("123_", new SseEmitter());
+    Map.Entry<String, SseEmitter> entry3 =
+        new AbstractMap.SimpleEntry<>("123_456", new SseEmitter());
+
+    long now = System.currentTimeMillis();
+    long threshold = 1000L;
+
+    assertThat(cleaner.isExpiredEmitterEntry(entry1, now, threshold)).isTrue();
+    assertThat(cleaner.isExpiredEmitterEntry(entry2, now, threshold)).isTrue();
+    assertThat(cleaner.isExpiredEmitterEntry(entry3, now, threshold)).isTrue();
+  }
+}
diff --git a/src/test/java/com/moa/moa_server/domain/notification/repository/FakeEmitterRepository.java b/src/test/java/com/moa/moa_server/domain/notification/repository/FakeEmitterRepository.java
@@ -0,0 +1,36 @@
+package com.moa.moa_server.domain.notification.repository;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+public class FakeEmitterRepository implements EmitterRepository {
+  private final Map<String, SseEmitter> store = new ConcurrentHashMap<>();
+
+  @Override
+  public void save(String id, SseEmitter emitter) {
+    store.put(id, emitter);
+  }
+
+  @Override
+  public void deleteById(String id) {
+    store.remove(id);
+  }
+
+  @Override
+  public Map<String, SseEmitter> findAllEmitters() {
+    return store;
+  }
+
+  @Override
+  public Map<String, SseEmitter> findAllByUserId(Long userId) {
+    return store.entrySet().stream()
+        .filter(e -> e.getKey().startsWith(userId + "_"))
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+  }
+
+  public boolean contains(String key) {
+    return store.containsKey(key);
+  }
+}
