diff --git a/src/main/java/com/debatetimer/service/sharing/SharingRoomRegistry.java b/src/main/java/com/debatetimer/service/sharing/SharingRoomRegistry.java index 159956d3..3ab58885 100644 --- a/src/main/java/com/debatetimer/service/sharing/SharingRoomRegistry.java +++ b/src/main/java/com/debatetimer/service/sharing/SharingRoomRegistry.java @@ -14,10 +14,13 @@ public class SharingRoomRegistry { static final Duration FINISHED_TTL = Duration.ofDays(1); static final Duration VERSION_TTL = Duration.ofDays(1); + static final Duration SYNC_REQUEST_TTL = Duration.ofDays(1); + static final Duration SYNC_REQUEST_INTERVAL = Duration.ofMillis(500); private static final int ROOM_LOCK_STRIPES = 64; private final Map finishedRoomExpirations = new ConcurrentHashMap<>(); private final Map lastAcceptedVersions = new ConcurrentHashMap<>(); + private final Map lastSyncRequestedAt = new ConcurrentHashMap<>(); private final Object[] roomLocks = createRoomLocks(); private final Clock clock; @@ -77,6 +80,33 @@ public void resetVersion(long roomId) { runExclusively(roomId, () -> lastAcceptedVersions.remove(roomId)); } + /** + * 룸별로 사회자에게 보내는 상태 공유 요청의 최소 간격을 강제한다. + * 여러 청중이 동시에 입장하거나 재구독하면 요청이 그만큼 늘어나고, 사회자의 응답은 룸 전체로 중계되므로 + * 메시지가 크게 증폭된다. 짧은 간격의 뒤따르는 요청은 직전 요청의 응답이 대신하므로 버린다. + * + * @return 요청을 보내도 되면 true + */ + public boolean tryAcquireSyncRequest(long roomId) { + Instant now = clock.instant(); + AtomicBoolean acquired = new AtomicBoolean(false); + lastSyncRequestedAt.compute(roomId, (id, lastRequested) -> { + if (lastRequested != null && now.isBefore(lastRequested.plus(SYNC_REQUEST_INTERVAL))) { + return lastRequested; + } + acquired.set(true); + return now; + }); + return acquired.get(); + } + + /** + * 룸의 상태 공유 요청 간격 기준을 지운다. 다음 요청은 간격과 무관하게 허용된다. + */ + public void resetSyncRequest(long roomId) { + lastSyncRequestedAt.remove(roomId); + } + /** * 같은 룸의 작업(버전 수락·룸 상태 변경·중계, 버전 초기화)을 하나씩 실행한다. * 인바운드 메시지는 여러 스레드에서 동시에 처리될 수 있어, 낮은 버전의 상태 변경이나 중계가 @@ -92,6 +122,8 @@ public void removeExpired() { finishedRoomExpirations.values().removeIf(this::isExpired); lastAcceptedVersions.values() .removeIf(accepted -> isExpired(accepted.acceptedAt().plus(VERSION_TTL))); + lastSyncRequestedAt.values() + .removeIf(requestedAt -> isExpired(requestedAt.plus(SYNC_REQUEST_TTL))); } private boolean isExpired(Instant expiration) { diff --git a/src/main/java/com/debatetimer/service/sharing/SharingService.java b/src/main/java/com/debatetimer/service/sharing/SharingService.java index eec10636..3bfedf20 100644 --- a/src/main/java/com/debatetimer/service/sharing/SharingService.java +++ b/src/main/java/com/debatetimer/service/sharing/SharingService.java @@ -70,11 +70,13 @@ public void share(long roomId, String chairmanSessionId, SharingRequest request) /** * 사회자의 구독은 공유 시작을 뜻하므로, 최신 사회자로 등록하고 이전에 종료된 룸이라도 다시 진행 상태로 되돌린다. + * 새 사회자는 직전 공유 요청의 응답을 보낸 적이 없으므로, 상태 공유 요청 간격 기준도 함께 지운다. */ public void startChairman(long roomId, String chairmanSessionId, String simpSessionId) { sharingRoomRegistry.runExclusively(roomId, () -> { if (claimChairman(roomId, chairmanSessionId, simpSessionId).isAccepted()) { sharingRoomRegistry.reopen(roomId); + sharingRoomRegistry.resetSyncRequest(roomId); } }); } @@ -84,6 +86,8 @@ public void startChairman(long roomId, String chairmanSessionId, String simpSess * - 종료된 룸이면 청중에게 종료를 알린다. * - 활성 사회자가 없으면 청중에게 사회자 부재를 알린다. * - 그 외에는 사회자에게 현재 상태 공유를 요청한다. + * 사회자의 응답은 룸 전체로 중계되므로, 짧은 간격의 뒤따르는 요청은 직전 요청의 응답이 대신한다. + * 그래서 룸별 최소 간격 안의 요청은 보내지 않는다. */ public void joinAudience(long roomId) { if (sharingRoomRegistry.isFinished(roomId)) { @@ -95,6 +99,9 @@ public void joinAudience(long roomId) { new SharingResponse(TimerEventType.CHAIRMAN_ABSENT)); return; } + if (!sharingRoomRegistry.tryAcquireSyncRequest(roomId)) { + return; + } messagingTemplate.convertAndSend(CHAIRMAN_CHANNEL_PREFIX + roomId, ChairmanSharingRequest.syncRequest(roomId)); } diff --git a/src/test/java/com/debatetimer/controller/sharing/SharingWebSocketControllerTest.java b/src/test/java/com/debatetimer/controller/sharing/SharingWebSocketControllerTest.java index 3d95acf3..fd47c12e 100644 --- a/src/test/java/com/debatetimer/controller/sharing/SharingWebSocketControllerTest.java +++ b/src/test/java/com/debatetimer/controller/sharing/SharingWebSocketControllerTest.java @@ -65,6 +65,7 @@ void startChairman() throws InterruptedException { void reopenRoom() { sharingRoomRegistry.reopen(ROOM_ID); sharingRoomRegistry.resetVersion(ROOM_ID); + sharingRoomRegistry.resetSyncRequest(ROOM_ID); chairmanSessionRegistry.remove(ROOM_ID); } diff --git a/src/test/java/com/debatetimer/event/sharing/RoomSubscribeInterceptorTest.java b/src/test/java/com/debatetimer/event/sharing/RoomSubscribeInterceptorTest.java index 470a1740..9fc3a84a 100644 --- a/src/test/java/com/debatetimer/event/sharing/RoomSubscribeInterceptorTest.java +++ b/src/test/java/com/debatetimer/event/sharing/RoomSubscribeInterceptorTest.java @@ -59,6 +59,7 @@ void setUpRoomOwner() { void reopenRoom() { sharingRoomRegistry.reopen(ROOM_ID); sharingRoomRegistry.resetVersion(ROOM_ID); + sharingRoomRegistry.resetSyncRequest(ROOM_ID); chairmanSessionRegistry.remove(ROOM_ID); } @@ -220,6 +221,41 @@ class AfterMessageHandled { } } + @Nested + class SyncRequestInterval { + + @Test + void 청중이_연달아_구독해도_사회자에게_정보공유_트리거를_한_번만_발송한다() throws Exception { + QueueFrameHandler chairmanHandler = new QueueFrameHandler<>( + ChairmanSharingRequest.class); + subscribeChairman(chairmanHandler); + + stompSession.subscribe("/room/" + ROOM_ID, new MessageFrameHandler<>(SharingResponse.class)); + assertThat(chairmanHandler.poll(3L)).isNotNull(); // 첫 구독 처리 완료 대기 + stompSession.subscribe("/room/" + ROOM_ID, new MessageFrameHandler<>(SharingResponse.class)); + + assertThat(chairmanHandler.poll(2L)).isNull(); + } + + @Test + void 요청_간격_기준을_지우면_사회자에게_정보공유_트리거를_다시_발송한다() throws Exception { + QueueFrameHandler chairmanHandler = new QueueFrameHandler<>( + ChairmanSharingRequest.class); + subscribeChairman(chairmanHandler); + stompSession.subscribe("/room/" + ROOM_ID, new MessageFrameHandler<>(SharingResponse.class)); + assertThat(chairmanHandler.poll(3L)).isNotNull(); // 첫 구독 처리 완료 대기 + + sharingRoomRegistry.resetSyncRequest(ROOM_ID); + stompSession.subscribe("/room/" + ROOM_ID, new MessageFrameHandler<>(SharingResponse.class)); + + ChairmanSharingRequest sharingRequest = chairmanHandler.poll(3L); + assertAll( + () -> assertThat(sharingRequest).isNotNull(), + () -> assertThat(sharingRequest.type()).isEqualTo(ChairmanNoticeType.SYNC_REQUEST) + ); + } + } + @Nested class ChairmanPresence { diff --git a/src/test/java/com/debatetimer/service/sharing/SharingRoomRegistryTest.java b/src/test/java/com/debatetimer/service/sharing/SharingRoomRegistryTest.java index e7c2b379..742f722a 100644 --- a/src/test/java/com/debatetimer/service/sharing/SharingRoomRegistryTest.java +++ b/src/test/java/com/debatetimer/service/sharing/SharingRoomRegistryTest.java @@ -266,6 +266,69 @@ class RemoveExpiredVersion { } } + @Nested + class TryAcquireSyncRequest { + + @Test + void 처음_요청은_허용한다() { + assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isTrue(); + } + + @Test + void 최소_간격_안의_뒤따르는_요청은_거절한다() { + sharingRoomRegistry.tryAcquireSyncRequest(1L); + + clock.advance(SharingRoomRegistry.SYNC_REQUEST_INTERVAL.minusMillis(1)); + + assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isFalse(); + } + + @Test + void 같은_시각의_연속_요청은_거절한다() { + sharingRoomRegistry.tryAcquireSyncRequest(1L); + + assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isFalse(); + } + + @Test + void 최소_간격이_지난_요청은_허용한다() { + sharingRoomRegistry.tryAcquireSyncRequest(1L); + + clock.advance(SharingRoomRegistry.SYNC_REQUEST_INTERVAL); + + assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isTrue(); + } + + @Test + void 거절된_요청은_마지막_요청_시각을_바꾸지_않는다() { + sharingRoomRegistry.tryAcquireSyncRequest(1L); + clock.advance(SharingRoomRegistry.SYNC_REQUEST_INTERVAL.minusMillis(1)); + sharingRoomRegistry.tryAcquireSyncRequest(1L); + + clock.advance(Duration.ofMillis(1)); + + assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isTrue(); + } + + @Test + void 룸마다_요청_간격을_따로_관리한다() { + sharingRoomRegistry.tryAcquireSyncRequest(1L); + + assertThat(sharingRoomRegistry.tryAcquireSyncRequest(2L)).isTrue(); + } + + @Test + void 유지_시간이_지난_요청_기록을_제거한다() { + sharingRoomRegistry.tryAcquireSyncRequest(1L); + clock.advance(SharingRoomRegistry.SYNC_REQUEST_TTL); + + sharingRoomRegistry.removeExpired(); + clock.advance(SharingRoomRegistry.SYNC_REQUEST_TTL.negated().plusMillis(1)); // 제거되지 않았다면 거절되는 시점 + + assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isTrue(); + } + } + private static class MutableClock extends Clock { private Instant instant; diff --git a/src/test/java/com/debatetimer/service/sharing/SharingServiceTest.java b/src/test/java/com/debatetimer/service/sharing/SharingServiceTest.java index dc705730..cf8d4d54 100644 --- a/src/test/java/com/debatetimer/service/sharing/SharingServiceTest.java +++ b/src/test/java/com/debatetimer/service/sharing/SharingServiceTest.java @@ -20,7 +20,9 @@ import com.debatetimer.dto.sharing.response.SharingResponse; import com.debatetimer.exception.custom.DTClientErrorException; import java.time.Clock; +import java.time.Duration; import java.time.Instant; +import java.time.ZoneId; import java.time.ZoneOffset; import java.util.List; import org.junit.jupiter.api.BeforeEach; @@ -37,6 +39,7 @@ class SharingServiceTest { private static final String CHAIRMAN = "tab-a"; private static final String SIMP_SESSION = "simp-1"; + private MutableClock clock; private SharingRoomRegistry sharingRoomRegistry; private ChairmanSessionRegistry chairmanSessionRegistry; private SimpMessageSendingOperations messagingTemplate; @@ -44,7 +47,7 @@ class SharingServiceTest { @BeforeEach void setUp() { - Clock clock = Clock.fixed(NOW, ZoneOffset.UTC); + clock = new MutableClock(NOW); sharingRoomRegistry = new SharingRoomRegistry(clock); chairmanSessionRegistry = new ChairmanSessionRegistry(clock); messagingTemplate = mock(SimpMessageSendingOperations.class); @@ -282,5 +285,56 @@ class JoinAudience { () -> verify(messagingTemplate, never()).convertAndSend(eq("/room/" + ROOM_ID), any(Object.class)) ); } + + @Test + void 최소_간격_안에_다시_입장하면_상태_공유를_요청하지_않는다() { + sharingService.startChairman(ROOM_ID, CHAIRMAN, SIMP_SESSION); + + sharingService.joinAudience(ROOM_ID); + clock.advance(SharingRoomRegistry.SYNC_REQUEST_INTERVAL.minusMillis(1)); + sharingService.joinAudience(ROOM_ID); + + assertThat(sentChairmanNotices(1)).hasSize(1); + } + + @Test + void 최소_간격이_지난_뒤_입장하면_상태_공유를_다시_요청한다() { + sharingService.startChairman(ROOM_ID, CHAIRMAN, SIMP_SESSION); + + sharingService.joinAudience(ROOM_ID); + clock.advance(SharingRoomRegistry.SYNC_REQUEST_INTERVAL); + sharingService.joinAudience(ROOM_ID); + + assertThat(sentChairmanNotices(2)) + .allMatch(notice -> notice.type() == ChairmanNoticeType.SYNC_REQUEST); + } + } + + private static class MutableClock extends Clock { + + private Instant instant; + + private MutableClock(Instant instant) { + this.instant = instant; + } + + private void advance(Duration duration) { + instant = instant.plus(duration); + } + + @Override + public ZoneId getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return instant; + } } }