diff --git a/src/main/java/kr/allcll/backend/config/ExternalPreInvoker.java b/src/main/java/kr/allcll/backend/config/ExternalPreInvoker.java index f7d359dc..585d168f 100644 --- a/src/main/java/kr/allcll/backend/config/ExternalPreInvoker.java +++ b/src/main/java/kr/allcll/backend/config/ExternalPreInvoker.java @@ -1,6 +1,7 @@ package kr.allcll.backend.config; import kr.allcll.backend.client.ExternalService; +import kr.allcll.backend.support.sse.SseEmitterStorage; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; @@ -12,6 +13,7 @@ public class ExternalPreInvoker { private final ExternalService externalService; + private final SseEmitterStorage sseEmitterStorage; @Scheduled(fixedDelay = 1000 * 10) void sendPinnedSubjectsToExternal() { @@ -23,4 +25,14 @@ void sendPinnedSubjectsToExternal() { } log.info("[ExternalPreInvoker] 핀 과목 전달 완료"); } + + @Scheduled(fixedDelay = 1000 * 60) + void cleanupExpiredSseActiveTimes() { + try { + sseEmitterStorage.cleanupExpiredActiveTimes(); + log.debug("[ExternalPreInvoker] SSE 활성 시각 정리 완료"); + } catch (Exception e) { + log.error("[ExternalPreInvoker] SSE 활성 시각 정리 중 오류 발생", e); + } + } } diff --git a/src/main/java/kr/allcll/backend/support/sse/SseConnection.java b/src/main/java/kr/allcll/backend/support/sse/SseConnection.java new file mode 100644 index 00000000..1d4df326 --- /dev/null +++ b/src/main/java/kr/allcll/backend/support/sse/SseConnection.java @@ -0,0 +1,39 @@ +package kr.allcll.backend.support.sse; + +import java.time.LocalDateTime; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +final class SseConnection { + + private SseEmitter emitter; + private final LocalDateTime lastActiveAt; + + SseConnection(SseEmitter emitter, LocalDateTime lastActiveAt) { + this.emitter = emitter; + this.lastActiveAt = lastActiveAt; + } + + static SseConnection of(SseEmitter emitter) { + return new SseConnection(emitter, LocalDateTime.now()); + } + + SseEmitter getEmitter() { + return emitter; + } + + boolean isConnected() { + return emitter != null; + } + + boolean isExpired(LocalDateTime threshold) { + return lastActiveAt.isBefore(threshold); + } + + boolean shouldCleanup(LocalDateTime threshold) { + return isExpired(threshold) && !isConnected(); + } + + void disconnect() { + this.emitter = null; + } +} diff --git a/src/main/java/kr/allcll/backend/support/sse/SseEmitterStorage.java b/src/main/java/kr/allcll/backend/support/sse/SseEmitterStorage.java index 645aa808..29795903 100644 --- a/src/main/java/kr/allcll/backend/support/sse/SseEmitterStorage.java +++ b/src/main/java/kr/allcll/backend/support/sse/SseEmitterStorage.java @@ -1,7 +1,10 @@ package kr.allcll.backend.support.sse; +import java.time.Duration; +import java.time.LocalDateTime; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import lombok.extern.slf4j.Slf4j; @@ -12,31 +15,33 @@ @Component public class SseEmitterStorage { - /* - ExecutorService로 변경이 필요할 수 있습니다. - */ - private final Map emitters; + private static final Duration GRACE_PERIOD = Duration.ofSeconds(30); + + private final Map connections; - /* - 동시성 문제로 자료구조 변경이 필요할 수 있습니다. - */ public SseEmitterStorage() { - this.emitters = new ConcurrentHashMap<>(); + this.connections = new ConcurrentHashMap<>(); } public void add(String token, SseEmitter sseEmitter) { - emitters.put(token, sseEmitter); - log.info("[SSE] 새로운 연결이 추가되었습니다. 현재 연결 수: {}", emitters.size()); + SseConnection connection = SseConnection.of(sseEmitter); + connections.put(token, connection); + log.info("[SSE] 새로운 연결이 추가되었습니다. 현재 연결 수: {}", getActiveConnectionCount()); + sseEmitter.onTimeout(() -> { - emitters.remove(token); - log.debug("[SSE] 연결이 타임아웃으로 종료되었습니다. 현재 연결 수: {}", emitters.size()); - }); - sseEmitter.onError(e -> { - emitters.remove(token); - }); - sseEmitter.onCompletion(() -> { - emitters.remove(token); + disconnectToken(token, sseEmitter); + log.debug("[SSE] 연결이 타임아웃으로 종료되었습니다. 현재 연결 수: {}", getActiveConnectionCount()); }); + sseEmitter.onError(e -> disconnectToken(token, sseEmitter)); + sseEmitter.onCompletion(() -> disconnectToken(token, sseEmitter)); + } + + private void disconnectToken(String token, SseEmitter sseEmitter) { + SseConnection connection = connections.get(token); + if (connection != null && connection.getEmitter() == sseEmitter) { + connection.disconnect(); + connections.remove(token); + } } /* @@ -44,19 +49,64 @@ public void add(String token, SseEmitter sseEmitter) { Collections.unmodifiableList(), Stream.toList()를 사용하면 랩핑한 객체를 반환하기에 위 문제가 발생한다. */ public List getEmitters() { - return emitters.values().stream().toList(); + return connections.values().stream() + .map(SseConnection::getEmitter) + .filter(Objects::nonNull) + .toList(); } public Optional getEmitter(String token) { - SseEmitter emitter = emitters.get(token); - return Optional.ofNullable(emitter); + SseConnection connection = connections.get(token); + if (connection == null) { + return Optional.empty(); + } + return Optional.ofNullable(connection.getEmitter()); } public List getUserTokens() { - return emitters.keySet().stream().toList(); + return getUserTokensWithinGracePeriod(); + } + + /** + * Grace Period 내에 활성화된 적이 있는 사용자 토큰 목록을 반환합니다. + * 현재 연결된 사용자와 최근 GRACE_PERIOD 이내에 연결이 끊긴 사용자를 포함합니다. + */ + private List getUserTokensWithinGracePeriod() { + LocalDateTime gracePeriodThreshold = LocalDateTime.now().minus(GRACE_PERIOD); + + return connections.entrySet().stream() + .filter(entry -> { + SseConnection sseConnection = entry.getValue(); + + if (sseConnection.isConnected()) { + return true; + } + + return !sseConnection.isExpired(gracePeriodThreshold); + }) + .map(Map.Entry::getKey) + .toList(); + } + + /** + * Grace Period를 초과한 오래된 연결 엔트리를 정리합니다. + * 주기적으로 호출되어 메모리 누수를 방지합니다. + */ + public void cleanupExpiredActiveTimes() { + LocalDateTime cleanupThreshold = LocalDateTime.now().minus(GRACE_PERIOD); + + connections.entrySet().removeIf(entry -> { + boolean remove = entry.getValue().shouldCleanup(cleanupThreshold); + if (remove) { + log.debug("[SSE] Grace Period 초과로 토큰 정리: {}", entry.getKey()); + } + return remove; + }); } public int getActiveConnectionCount() { - return emitters.size(); + return (int) connections.values().stream() + .filter(SseConnection::isConnected) + .count(); } } diff --git a/src/test/java/kr/allcll/backend/support/sse/SseEmitterStorageTest.java b/src/test/java/kr/allcll/backend/support/sse/SseEmitterStorageTest.java new file mode 100644 index 00000000..c5f1dcb0 --- /dev/null +++ b/src/test/java/kr/allcll/backend/support/sse/SseEmitterStorageTest.java @@ -0,0 +1,330 @@ +package kr.allcll.backend.support.sse; + + +import static org.assertj.core.api.Assertions.assertThat; + +import org.assertj.core.api.SoftAssertions; + +import java.lang.reflect.Field; +import java.time.LocalDateTime; +import java.util.List; +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; + +class SseEmitterStorageTest { + + private SseEmitterStorage storage; + + @BeforeEach + void setUp() { + storage = new SseEmitterStorage(); + } + + @DisplayName("SSE 연결 추가 시 getUserTokens에서 토큰이 조회된다") + @Test + void add_shouldRecordLastActiveTime() { + // given + String token = "test-token"; + SseEmitter emitter = new SseEmitter(); + + // when + storage.add(token, emitter); + + // then + List tokens = storage.getUserTokens(); + assertThat(tokens).contains(token); + } + + @DisplayName("여러 연결이 있을 때 모두 토큰으로 조회된다") + @Test + void getUserTokens_shouldReturnAllTokens() { + // given + String token1 = "token1"; + String token2 = "token2"; + String token3 = "token3"; + + SseEmitter emitter1 = new SseEmitter(); + SseEmitter emitter2 = new SseEmitter(); + SseEmitter emitter3 = new SseEmitter(); + + storage.add(token1, emitter1); + storage.add(token2, emitter2); + storage.add(token3, emitter3); + + // when + List tokens = storage.getUserTokens(); + + // then + assertThat(tokens).containsExactlyInAnyOrder(token1, token2, token3); + } + + @DisplayName("정리 작업 실행 시 최근 토큰은 제거되지 않는다") + @Test + void cleanupExpiredActiveTimes_shouldNotRemoveRecentTokens() { + // given + String token = "test-token"; + SseEmitter emitter = new SseEmitter(); + storage.add(token, emitter); + + // when + storage.cleanupExpiredActiveTimes(); + + // then + List tokens = storage.getUserTokens(); + assertThat(tokens).contains(token); + } + + @DisplayName("활성 연결 수는 실제 emitters 수와 동일하다") + @Test + void getActiveConnectionCount_shouldReturnEmittersSize() { + // given + String token1 = "token1"; + String token2 = "token2"; + + SseEmitter emitter1 = new SseEmitter(); + SseEmitter emitter2 = new SseEmitter(); + + storage.add(token1, emitter1); + storage.add(token2, emitter2); + + // when & then + assertThat(storage.getActiveConnectionCount()).isEqualTo(2); + } + + @DisplayName("Emitter를 정상적으로 가져올 수 있다") + @Test + void getEmitter_shouldReturnCorrectEmitter() { + // given + String token = "test-token"; + SseEmitter emitter = new SseEmitter(); + storage.add(token, emitter); + + // when & then + assertThat(storage.getEmitter(token)).isPresent(); + } + + @DisplayName("존재하지 않는 토큰으로 조회 시 빈 Optional을 반환한다") + @Test + void getEmitter_shouldReturnEmptyForNonExistentToken() { + // when & then + assertThat(storage.getEmitter("non-existent")).isEmpty(); + } + + @DisplayName("30초 이상 유지되는 정상 SSE 연결이 활성으로 처리되어야 한다") + @Test + void getUserTokens_shouldNotExcludeLongLivedActiveConnections() throws Exception { + // given + String token = "long-lived-connection"; + SseEmitter emitter = new SseEmitter(60000L); + storage.add(token, emitter); + + // when + SoftAssertions.assertSoftly(softly -> { + softly.assertThat(storage.getUserTokens()).contains(token); + softly.assertThat(storage.getActiveConnectionCount()).isEqualTo(1); + softly.assertThat(storage.getEmitter(token)).isPresent(); + }); + + setLastActiveTime(token, LocalDateTime.now().minusSeconds(31)); + + // then + assertThat(storage.getActiveConnectionCount()).isEqualTo(1); + assertThat(storage.getEmitter(token)).isPresent(); + + List activeTokens = storage.getUserTokens(); + + assertThat(activeTokens) + .contains(token); + } + + @DisplayName("emitters에 존재하는 연결은 항상 getUserTokens에 포함되어야 한다") + @Test + void getUserTokens_shouldIncludeAllActiveEmitters() throws Exception { + // given + String oldToken = "old-active-connection"; + String newToken = "new-active-connection"; + + SseEmitter oldEmitter = new SseEmitter(60000L); + SseEmitter newEmitter = new SseEmitter(60000L); + + storage.add(oldToken, oldEmitter); + + // oldToken의 lastActiveTime을 31초 전으로 설정 + setLastActiveTime(oldToken, LocalDateTime.now().minusSeconds(31)); + storage.add(newToken, newEmitter); + + // when + List activeTokens = storage.getUserTokens(); + int activeCount = storage.getActiveConnectionCount(); + + // then + assertThat(activeCount).isEqualTo(2); + + assertThat(activeTokens) + .containsExactlyInAnyOrder(oldToken, newToken); + } + + @DisplayName("연결 종료 후 30초 이내에는 getUserTokens에 포함되어야 한다 (Grace Period)") + @Test + void getUserTokens_shouldIncludeRecentlyDisconnectedTokensWithinGracePeriod() throws Exception { + // given + String token = "disconnected-token"; + SseEmitter emitter = new SseEmitter(); + storage.add(token, emitter); + + // when + removeFromEmitters(token); + + SoftAssertions.assertSoftly(softly -> { + softly.assertThat(storage.getActiveConnectionCount()).isEqualTo(0); + softly.assertThat(storage.getEmitter(token)).isEmpty(); + softly.assertThat(storage.getUserTokens()).contains(token); + }); + + setLastActiveTime(token, LocalDateTime.now().minusSeconds(31)); + + assertThat(storage.getUserTokens()).doesNotContain(token); + } + + @DisplayName("cleanupExpiredActiveTimes는 Grace Period를 초과한 비활성 토큰만 제거한다") + @Test + void cleanupExpiredActiveTimes_shouldOnlyRemoveExpiredInactiveTokens() throws Exception { + // given + /// 활성 연결 추가 + String activeToken = "still-connected"; + SseEmitter activeEmitter = new SseEmitter(60000L); + storage.add(activeToken, activeEmitter); + + /// 만료될 비활성 연결 추가 + String expiredToken = "expired-disconnected"; + SseEmitter expiredEmitter = new SseEmitter(); + storage.add(expiredToken, expiredEmitter); + removeFromEmitters(expiredToken); + setLastActiveTime(expiredToken, LocalDateTime.now().minusSeconds(31)); + + // when + storage.cleanupExpiredActiveTimes(); + + // then + List tokens = storage.getUserTokens(); + assertThat(tokens) + .contains(activeToken) + .doesNotContain(expiredToken); + } + + @DisplayName("같은 토큰으로 재연결 시 이전 emitter의 콜백이 새 연결을 끊지 않아야 한다") + @Test + void reconnection_shouldNotDisconnectNewEmitterByOldEmitterCallback() { + // given + String token = "reconnect-user"; + SseEmitter oldEmitter = new SseEmitter(5000L); + storage.add(token, oldEmitter); + + // when + SseEmitter newEmitter = new SseEmitter(60000L); + storage.add(token, newEmitter); + oldEmitter.complete(); + + // then + SoftAssertions.assertSoftly(softly -> { + softly.assertThat(storage.getEmitter(token)) + .isPresent(); + softly.assertThat(storage.getActiveConnectionCount()) + .isEqualTo(1); + softly.assertThat(storage.getUserTokens()) + .contains(token); + }); + } + + @DisplayName("재연결 시 이전 연결의 여러 콜백이 동시에 실행되어도 새 연결이 유지되어야 한다") + @Test + void reconnection_shouldMaintainNewConnectionEvenWithMultipleOldCallbacks() { + // given + String token = "multi-callback-user"; + SseEmitter oldEmitter = new SseEmitter(5000L); + storage.add(token, oldEmitter); + + SseEmitter newEmitter = new SseEmitter(60000L); + storage.add(token, newEmitter); + + assertThat(storage.getActiveConnectionCount()).isEqualTo(1); + + // when + oldEmitter.completeWithError(new RuntimeException("Old connection error")); + oldEmitter.complete(); + + // then + assertThat(storage.getEmitter(token)).isPresent(); + assertThat(storage.getActiveConnectionCount()).isEqualTo(1); + assertThat(storage.getUserTokens()).contains(token); + } + + @DisplayName("동일 토큰으로 빠르게 여러 번 재연결해도 마지막 연결만 유지되어야 한다") + @Test + void reconnection_shouldKeepOnlyLastConnectionAfterMultipleReconnects() { + // given + String token = "rapid-reconnect-user"; + + // when + SseEmitter emitter1 = new SseEmitter(); + storage.add(token, emitter1); + assertThat(storage.getActiveConnectionCount()).isEqualTo(1); + + SseEmitter emitter2 = new SseEmitter(); + storage.add(token, emitter2); + assertThat(storage.getActiveConnectionCount()).isEqualTo(1); + + SseEmitter emitter3 = new SseEmitter(); + storage.add(token, emitter3); + + // 이전 emitter들의 콜백 실행 + emitter1.complete(); + emitter2.complete(); + + // then - 마지막 연결만 활성 상태 + assertThat(storage.getActiveConnectionCount()).isEqualTo(1); + assertThat(storage.getEmitter(token)).isPresent(); + assertThat(storage.getUserTokens()).contains(token); + } + + @SuppressWarnings("unchecked") + private void setLastActiveTime(String token, LocalDateTime time) throws Exception { + Field connectionsField = SseEmitterStorage.class.getDeclaredField("connections"); + connectionsField.setAccessible(true); + Map connections = (Map) connectionsField.get(storage); + + Object connection = connections.get(token); + if (connection != null) { + Field emitterField = connection.getClass().getDeclaredField("emitter"); + emitterField.setAccessible(true); + SseEmitter emitter = (SseEmitter) emitterField.get(connection); + + Class sseConnectionClass = Class.forName("kr.allcll.backend.support.sse.SseConnection"); + var constructor = sseConnectionClass.getDeclaredConstructor(SseEmitter.class, LocalDateTime.class); + constructor.setAccessible(true); + Object newConnection = constructor.newInstance(emitter, time); + connections.put(token, newConnection); + } + } + + /** + - 연결 종료 시뮬레이션 (onCompletion 콜백을 수동으로 실행) + - 실제 HTTP 연결 없는 테스트 환경에서는 complete() 호출이 콜백을 트리거하지 않음 + **/ + @SuppressWarnings("unchecked") + private void removeFromEmitters(String token) throws Exception { + Field connectionsField = SseEmitterStorage.class.getDeclaredField("connections"); + connectionsField.setAccessible(true); + Map connections = (Map) connectionsField.get(storage); + + Object connection = connections.get(token); + if (connection != null) { + var disconnectMethod = connection.getClass().getDeclaredMethod("disconnect"); + disconnectMethod.setAccessible(true); + disconnectMethod.invoke(connection); + } + } +}