fix(dm): 실시간 연결 상태 처리를 보강한다

app/src/main/java/kr/co/vividnext/sodalive/v2/main/chat/dm/DmChatRoomViewModel.kt

@@ -9,6 +9,8 @@ import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
 import io.reactivex.rxjava3.core.Scheduler
 import io.reactivex.rxjava3.disposables.Disposable
 import io.reactivex.rxjava3.schedulers.Schedulers
+import kr.co.vividnext.sodalive.R
+import kr.co.vividnext.sodalive.common.SodaLiveApplicationHolder
 import kr.co.vividnext.sodalive.base.BaseViewModel
 import kr.co.vividnext.sodalive.common.ApiResponse
 import kr.co.vividnext.sodalive.common.SharedPreferenceManager
@@ -26,6 +28,7 @@ import kr.co.vividnext.sodalive.v2.main.chat.dm.model.mergeByMessageId
 import kr.co.vividnext.sodalive.v2.main.chat.dm.model.sortByCreatedAtAndMessageId
 import kr.co.vividnext.sodalive.v2.main.chat.dm.model.toUiItem
 import kr.co.vividnext.sodalive.v2.main.chat.dm.model.toUiItems
+import java.util.UUID
 import java.util.concurrent.TimeUnit
 
 class DmChatRoomViewModel(
@@ -43,6 +46,7 @@ class DmChatRoomViewModel(
     private var isLoadingOlder: Boolean = false
     private var isRealtimeJoining: Boolean = false
     private var isRealtimeConnected: Boolean = false
+    private var joinErrorRetryCount: Int = 0
     private var shouldReconnectRealtime: Boolean = false
     private var currentAuthToken: String = ""
     private var currentRealtimeToken: String = ""
@@ -51,7 +55,6 @@ class DmChatRoomViewModel(
     private var heartbeatPingDisposable: Disposable? = null
     private var heartbeatTimeoutDisposable: Disposable? = null
     private var localMessageSequence: Long = 0L
-    private var requestSequence: Long = 0L
     private val pendingRequestLocalIds = mutableMapOf<String, String>()
     private val recentFailedRequestLocalIds = mutableMapOf<String, String>()
     private val pendingTimeoutDisposables = mutableMapOf<String, Disposable>()
@@ -73,6 +76,10 @@ class DmChatRoomViewModel(
     val roomOpenedEventLiveData: LiveData<DmChatEvent<Boolean>>
         get() = _roomOpenedEventLiveData
 
+    private val _isRealtimeConnectingLiveData = MutableLiveData(false)
+    val isRealtimeConnectingLiveData: LiveData<Boolean>
+        get() = _isRealtimeConnectingLiveData
+
     fun enter(roomId: Long, creatorId: Long) {
         when {
             roomId > 0L -> openRoom(roomId)
@@ -108,9 +115,9 @@ class DmChatRoomViewModel(
         )
     }
 
-    fun sendText(text: String) {
+    fun sendText(text: String): Boolean {
         val trimmed = text.trim()
-        if (trimmed.isBlank() || currentRoomId <= 0L) return
+        if (trimmed.isBlank() || currentRoomId <= 0L || !isRealtimeConnected) return false
 
         val localId = nextLocalId()
         val requestId = nextRequestId()
@@ -131,13 +138,14 @@ class DmChatRoomViewModel(
         emitContent()
 
         sendLocalMessage(requestId = requestId, text = trimmed)
+        return true
     }
 
     fun retry(localId: String) {
         val failedItem = currentMessages.firstOrNull {
             it.localId == localId && it.status == DmChatMessageStatus.FAILED
         } ?: return
-        if (currentRoomId <= 0L) return
+        if (currentRoomId <= 0L || !isRealtimeConnected) return
         val requestId = nextRequestId()
 
         removeRecentFailedRequests(localId)
@@ -207,6 +215,7 @@ class DmChatRoomViewModel(
         currentRealtimeRoomId = roomId
         isRealtimeJoining = true
         isRealtimeConnected = false
+        setRealtimeConnecting(true)
         shouldReconnectRealtime = true
         reconnectDisposable?.dispose()
         reconnectDisposable = null
@@ -219,15 +228,12 @@ class DmChatRoomViewModel(
 
                 override fun onFailure(throwable: Throwable) {
                     scheduleRealtimeCallback {
-                        isRealtimeJoining = false
+                        handleRealtimeDisconnect(throwable)
-                        isRealtimeConnected = false
-                        throwable.message?.let { Logger.e(it) }
-                        scheduleRealtimeReconnect()
                     }
                 }
             }
         )
-        repository.sendJoinRoom(roomId)
+        repository.sendJoinRoom(roomId = roomId, requestId = nextRequestId())
     }
 
     fun leaveRealtime() {
@@ -237,15 +243,20 @@ class DmChatRoomViewModel(
             (isRealtimeJoining || isRealtimeConnected || currentRealtimeToken.isNotEmpty())
         if (!hasActiveSocket) return
 
+        val shouldSendLeave = isRealtimeConnected
         shouldReconnectRealtime = false
         stopHeartbeat()
         currentRealtimeToken = ""
         currentRealtimeRoomId = 0L
         isRealtimeJoining = false
         isRealtimeConnected = false
+        joinErrorRetryCount = 0
+        setRealtimeConnecting(false)
         reconnectDisposable?.dispose()
         reconnectDisposable = null
-        repository.sendLeaveRoom(roomId)
+        if (shouldSendLeave) {
+            repository.sendLeaveRoom(roomId = roomId, requestId = nextRequestId())
+        }
         repository.closeSocket()
     }
 
@@ -282,6 +293,8 @@ class DmChatRoomViewModel(
         currentRealtimeRoomId = 0L
         isRealtimeJoining = false
         isRealtimeConnected = false
+        joinErrorRetryCount = 0
+        setRealtimeConnecting(false)
         repository.closeSocket()
         super.onCleared()
     }
@@ -344,6 +357,7 @@ class DmChatRoomViewModel(
             return
         }
 
+        joinErrorRetryCount = 0
         currentRoomId = data.roomId
         opponentNickname = data.opponentNickname
         opponentProfileImageUrl = data.opponentProfileImageUrl
@@ -375,17 +389,70 @@ class DmChatRoomViewModel(
 
     private fun handleSocketEvent(event: DmChatSocketEvent, token: String) {
         when (event) {
-            DmChatSocketEvent.Joined -> {
+            is DmChatSocketEvent.Joined -> {
+                if (!isCurrentRoomEvent(event.roomId)) return
+                joinErrorRetryCount = 0
                 isRealtimeJoining = false
                 isRealtimeConnected = true
+                setRealtimeConnecting(false)
                 startHeartbeat()
                 syncLatestMessagesAfterReconnect(token = token)
             }
-            is DmChatSocketEvent.Message -> handleRealtimeMessage(event.requestId, event.message)
+            is DmChatSocketEvent.Message -> {
-            is DmChatSocketEvent.SendAck -> handleSendAck(event.requestId, event.message)
+                if (!isRealtimeConnected || !isStrictCurrentRoomEvent(event.roomId)) return
-            is DmChatSocketEvent.Error -> event.requestId?.let { markPendingMessageFailed(it) }
+                handleRealtimeMessage(event.requestId, event.message)
-            DmChatSocketEvent.Pong -> clearHeartbeatTimeout()
             }
+            is DmChatSocketEvent.SendAck -> {
+                if (!isRealtimeConnected || !isStrictCurrentRoomEvent(event.roomId)) return
+                handleSendAck(event.requestId, event.message)
+            }
+            is DmChatSocketEvent.Error -> {
+                if (isRealtimeJoining && !isRealtimeConnected) {
+                    handleJoinErrorBeforeJoined()
+                    return
+                }
+                if (!isRealtimeConnected || !isCurrentRoomEvent(event.roomId)) return
+                event.requestId?.let { markPendingMessageFailed(it) }
+            }
+            is DmChatSocketEvent.Pong -> {
+                if (!isRealtimeConnected || !isCurrentRoomEvent(event.roomId)) return
+                clearHeartbeatTimeout()
+            }
+        }
+    }
+
+    private fun isCurrentRoomEvent(roomId: Long?): Boolean = roomId == null || roomId == currentRoomId
+
+    private fun isStrictCurrentRoomEvent(roomId: Long?): Boolean = roomId == currentRoomId
+
+    private fun handleJoinErrorBeforeJoined() {
+        isRealtimeJoining = false
+        isRealtimeConnected = false
+        stopHeartbeat()
+        repository.closeSocket()
+        if (joinErrorRetryCount < MAX_JOIN_ERROR_RETRY_COUNT) {
+            joinErrorRetryCount += 1
+            setRealtimeConnecting(true)
+            scheduleRealtimeReconnect()
+            return
+        }
+
+        failRealtimeRoomEntry()
+    }
+
+    private fun failRealtimeRoomEntry() {
+        shouldReconnectRealtime = false
+        joinErrorRetryCount = 0
+        reconnectDisposable?.dispose()
+        reconnectDisposable = null
+        currentRealtimeToken = ""
+        currentRealtimeRoomId = 0L
+        isRealtimeJoining = false
+        isRealtimeConnected = false
+        setRealtimeConnecting(false)
+        repository.closeSocket()
+        showError(SodaLiveApplicationHolder.get().getString(R.string.screen_dm_chat_room_enter_failed))
+        _finishEventLiveData.value = true
     }
 
     private fun startHeartbeat() {
@@ -399,7 +466,11 @@ class DmChatRoomViewModel(
                         connectRealtime(token = latestToken)
                         return@scheduleRealtimeCallback
                     }
-                    if (repository.sendPing()) scheduleHeartbeatTimeout()
+                    if (repository.sendPing(roomId = currentRoomId, requestId = nextRequestId())) {
+                        scheduleHeartbeatTimeout()
+                    } else {
+                        handleRealtimeDisconnect(IllegalStateException("WebSocket PING send failed"))
+                    }
                 }
             },
             HEARTBEAT_INTERVAL_MILLIS,
@@ -416,6 +487,7 @@ class DmChatRoomViewModel(
                     if (!isRealtimeConnected || !shouldReconnectRealtime) return@scheduleRealtimeCallback
                     isRealtimeJoining = false
                     isRealtimeConnected = false
+                    setRealtimeConnecting(false)
                     stopHeartbeat()
                     repository.closeSocket()
                     scheduleRealtimeReconnect()
@@ -426,6 +498,16 @@ class DmChatRoomViewModel(
         ).also { compositeDisposable.add(it) }
     }
 
+    private fun handleRealtimeDisconnect(throwable: Throwable) {
+        isRealtimeJoining = false
+        isRealtimeConnected = false
+        setRealtimeConnecting(false)
+        stopHeartbeat()
+        repository.closeSocket()
+        throwable.message?.let { Logger.e(it) }
+        scheduleRealtimeReconnect()
+    }
+
     private fun stopHeartbeat() {
         heartbeatPingDisposable?.dispose()
         heartbeatPingDisposable = null
@@ -437,6 +519,11 @@ class DmChatRoomViewModel(
         heartbeatTimeoutDisposable = null
     }
 
+    private fun setRealtimeConnecting(isConnecting: Boolean) {
+        if (_isRealtimeConnectingLiveData.value == isConnecting) return
+        _isRealtimeConnectingLiveData.value = isConnecting
+    }
+
     private fun handleSendAck(requestId: String, message: DmChatMessageResponse) {
         val localId = pendingRequestLocalIds.remove(requestId)
             ?: recentFailedRequestLocalIds.remove(requestId)
@@ -522,10 +609,7 @@ class DmChatRoomViewModel(
         return "local-$localMessageSequence"
     }
 
-    private fun nextRequestId(): String {
+    private fun nextRequestId(): String = UUID.randomUUID().toString()
-        requestSequence += 1L
-        return "request-$requestSequence"
-    }
 
     private fun authToken(): String {
         val token = tokenProvider()
@@ -540,9 +624,10 @@ class DmChatRoomViewModel(
 
     private companion object {
         const val RECONNECT_DELAY_MILLIS = 3_000L
-        const val SEND_ACK_TIMEOUT_MILLIS = 10_000L
+        const val SEND_ACK_TIMEOUT_MILLIS = 15_000L
         const val HEARTBEAT_INTERVAL_MILLIS = 30_000L
         const val HEARTBEAT_TIMEOUT_MILLIS = 10_000L
+        const val MAX_JOIN_ERROR_RETRY_COUNT = 3
     }
 }
 

app/src/test/java/kr/co/vividnext/sodalive/v2/main/chat/dm/DmChatRoomViewModelTest.kt

@@ -14,8 +14,10 @@ import io.reactivex.rxjava3.core.Single
 import io.reactivex.rxjava3.plugins.RxJavaPlugins
 import io.reactivex.rxjava3.schedulers.Schedulers
 import io.reactivex.rxjava3.schedulers.TestScheduler
+import kr.co.vividnext.sodalive.R
 import kr.co.vividnext.sodalive.common.ApiResponse
 import kr.co.vividnext.sodalive.common.SharedPreferenceManager
+import kr.co.vividnext.sodalive.common.SodaLiveApplicationHolder
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.CreateDmChatRoomRequest
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.CreateDmChatRoomResponse
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatApi
@@ -24,9 +26,12 @@ import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatMessagesPageResponse
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatRepository
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatRoomOpenResponse
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatSocketClient
+import kr.co.vividnext.sodalive.v2.main.chat.dm.data.SendDmVoiceMessageResponse
 import kr.co.vividnext.sodalive.v2.main.chat.dm.model.DmChatMessageStatus
 import kr.co.vividnext.sodalive.v2.main.chat.dm.model.DmChatRoomUiState
+import okhttp3.MultipartBody
 import okhttp3.OkHttpClient
+import okhttp3.RequestBody
 import okhttp3.Request
 import okhttp3.WebSocket
 import okhttp3.WebSocketListener
@@ -62,6 +67,7 @@ class DmChatRoomViewModelTest {
         setImmediateRxSchedulers()
         SharedPreferenceManager.resetForTest()
         SharedPreferenceManager.init(context)
+        SodaLiveApplicationHolder.init(context as Application)
         SharedPreferenceManager.token = "test-token"
         token = "test-token"
         api = FakeDmChatApi()
@@ -159,25 +165,147 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
 
-        viewModel.sendText("   ")
+        val accepted = viewModel.sendText("   ")
 
+        assertEquals(false, accepted)
         assertTrue(socketFactory.webSocket.sentTexts.isEmpty())
         val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         assertTrue(state.messages.isEmpty())
     }
 
+    @Test
+    fun `JOINED 전에는 텍스트를 전송하지 않는다`() {
+        api.enqueueOpenSuccess(openResponse(roomId = 10L))
+        viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
+
+        val accepted = viewModel.sendText("안녕")
+
+        assertEquals(false, accepted)
+        assertEquals(listOf("JOIN_ROOM"), socketFactory.webSocket.sentTexts.map { it.type() })
+        val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
+        assertTrue(state.messages.isEmpty())
+    }
+
+    @Test
+    fun `텍스트 전송 requestId는 UUID 형식이다`() {
+        api.enqueueOpenSuccess(openResponse(roomId = 10L))
+        viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
+        socketFactory.emitJoined()
+
+        val accepted = viewModel.sendText("안녕")
+
+        assertEquals(true, accepted)
+        val requestId = (viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content)
+            .messages.single().requestId!!
+        assertTrue(requestId.matches(Regex("[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}")))
+    }
+
+    @Test
+    fun `pending timeout은 15초 기준으로 실패 처리한다`() {
+        api.enqueueOpenSuccess(openResponse(roomId = 10L))
+        viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
+        socketFactory.emitJoined()
+        viewModel.sendText("timeout")
+
+        reconnectScheduler.advanceTimeBy(10L, TimeUnit.SECONDS)
+        shadowOf(Looper.getMainLooper()).idle()
+        var state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
+        assertEquals(DmChatMessageStatus.SENDING, state.messages.single().status)
+
+        reconnectScheduler.advanceTimeBy(5L, TimeUnit.SECONDS)
+        shadowOf(Looper.getMainLooper()).idle()
+        state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
+        assertEquals(DmChatMessageStatus.FAILED, state.messages.single().status)
+    }
+
+    @Test
+    fun `현재 roomId와 다른 socket MESSAGE와 SEND_ACK는 무시한다`() {
+        api.enqueueOpenSuccess(openResponse(roomId = 10L))
+        viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
+        socketFactory.emitJoined()
+        viewModel.sendText("안녕")
+        val requestId = (viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content)
+            .messages.single().requestId!!
+
+        socketFactory.emitMessage(roomId = 11L, message = message(messageId = 80L, textMessage = "다른 방"))
+        socketFactory.emitAck(
+            roomId = 11L,
+            requestId = requestId,
+            message = message(messageId = 81L, mine = true, textMessage = "다른 ACK")
+        )
+
+        val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
+        assertEquals(listOf(null), state.messages.map { it.messageId })
+        assertEquals(listOf("안녕"), state.messages.map { it.textMessage })
+        assertEquals(listOf(DmChatMessageStatus.SENDING), state.messages.map { it.status })
+    }
+
+    @Test
+    fun `roomId가 null인 socket MESSAGE와 SEND_ACK는 무시한다`() {
+        api.enqueueOpenSuccess(openResponse(roomId = 10L))
+        viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
+        socketFactory.emitJoined()
+        viewModel.sendText("안녕")
+        val requestId = (viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content)
+            .messages.single().requestId!!
+
+        socketFactory.emitMessage(roomId = null, message = message(messageId = 80L, textMessage = "malformed"))
+        socketFactory.emitAck(
+            roomId = null,
+            requestId = requestId,
+            message = message(messageId = 81L, mine = true, textMessage = "malformed ACK")
+        )
+
+        val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
+        assertEquals(listOf(null), state.messages.map { it.messageId })
+        assertEquals(listOf(DmChatMessageStatus.SENDING), state.messages.map { it.status })
+    }
+
+    @Test
+    fun `JOINED 전 ERROR는 세 번 재시도 후 로딩 해제 오류 표시 종료를 수행한다`() {
+        api.enqueueOpenSuccess(openResponse(roomId = 10L))
+        viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
+
+        repeat(3) { retryIndex ->
+            socketFactory.emitJoinError(roomId = 0L)
+            assertEquals(true, viewModel.isRealtimeConnectingLiveData.requireValue())
+            reconnectScheduler.advanceTimeBy(3L, TimeUnit.SECONDS)
+            shadowOf(Looper.getMainLooper()).idle()
+            assertEquals(retryIndex + 2, socketFactory.connectCalls.size)
+        }
+
+        socketFactory.emitJoinError(roomId = 0L)
+
+        assertEquals(false, viewModel.isRealtimeConnectingLiveData.requireValue())
+        assertEquals(true, viewModel.finishEventLiveData.requireValue())
+        val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Error
+        assertEquals(context.getString(R.string.screen_dm_chat_room_enter_failed), state.message)
+        val connectCountAfterFailure = socketFactory.connectCalls.size
+        reconnectScheduler.advanceTimeBy(3L, TimeUnit.SECONDS)
+        shadowOf(Looper.getMainLooper()).idle()
+        assertEquals(connectCountAfterFailure, socketFactory.connectCalls.size)
+        assertTrue(socketFactory.closeCount >= 4)
+    }
+
     @Test
     fun `전송 직후 pending을 추가하고 성공 시 서버 메시지로 교체한다`() {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
 
         viewModel.sendText(" 안녕 ")
         val sendingState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         val requestId = sendingState.messages.single().requestId!!
 
         assertEquals("SEND_TEXT", socketFactory.webSocket.sentJsonAt(1).get("type").asString)
-        assertEquals(requestId, socketFactory.webSocket.sentJsonAt(1).getAsJsonObject("payload").get("requestId").asString)
+        assertEquals(requestId, socketFactory.webSocket.sentJsonAt(1).get("requestId").asString)
         assertEquals(DmChatMessageStatus.SENDING, sendingState.messages.single().status)
         assertEquals("안녕", sendingState.messages.single().textMessage)
 
@@ -192,6 +320,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
 
         viewModel.sendText("안녕")
         viewModel.sendText("안녕")
@@ -206,6 +335,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
 
         viewModel.sendText("첫번째")
         viewModel.sendText("두번째")
@@ -217,8 +347,8 @@ class DmChatRoomViewModelTest {
         assertNotEquals(requestIds[0], requestIds[1])
         assertEquals("SEND_TEXT", socketFactory.webSocket.sentJsonAt(1).get("type").asString)
         assertEquals("SEND_TEXT", socketFactory.webSocket.sentJsonAt(2).get("type").asString)
-        assertEquals(requestIds[0], socketFactory.webSocket.sentJsonAt(1).getAsJsonObject("payload").get("requestId").asString)
+        assertEquals(requestIds[0], socketFactory.webSocket.sentJsonAt(1).get("requestId").asString)
-        assertEquals(requestIds[1], socketFactory.webSocket.sentJsonAt(2).getAsJsonObject("payload").get("requestId").asString)
+        assertEquals(requestIds[1], socketFactory.webSocket.sentJsonAt(2).get("requestId").asString)
     }
 
     @Test
@@ -226,6 +356,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         viewModel.sendText("첫번째")
         viewModel.sendText("두번째")
         val pendingState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
@@ -244,6 +375,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         viewModel.sendText("첫번째")
         viewModel.sendText("두번째")
         val pendingState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
@@ -260,13 +392,14 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         viewModel.sendText("첫번째")
         viewModel.sendText("두번째")
         val pendingState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         val secondRequestId = pendingState.messages[1].requestId!!
 
         socketFactory.emitAck(secondRequestId, message(messageId = 31L, mine = true, textMessage = "두번째"))
-        reconnectScheduler.advanceTimeBy(10L, TimeUnit.SECONDS)
+        reconnectScheduler.advanceTimeBy(15L, TimeUnit.SECONDS)
         shadowOf(Looper.getMainLooper()).idle()
 
         val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
@@ -279,6 +412,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         viewModel.sendText("안녕")
         val requestId = (viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content)
             .messages.single().requestId!!
@@ -296,6 +430,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         viewModel.sendText("재시도")
         val failedRequestId = (viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content)
             .messages.single().requestId!!
@@ -311,7 +446,7 @@ class DmChatRoomViewModelTest {
         assertEquals(DmChatMessageStatus.SENDING, retryItem.status)
         assertEquals(
             retryItem.requestId,
-            socketFactory.webSocket.sentJsonAt(2).getAsJsonObject("payload").get("requestId").asString
+            socketFactory.webSocket.sentJsonAt(2).get("requestId").asString
         )
     }
 
@@ -320,6 +455,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         socketFactory.webSocket.sendResult = false
 
         viewModel.sendText("안녕")
@@ -343,6 +479,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         socketFactory.webSocket.sendResult = false
         viewModel.sendText("안녕")
         val failedState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
@@ -378,6 +515,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
 
         viewModel.sendText("안녕")
         val requestId = (viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content)
@@ -395,6 +533,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         viewModel.sendText("선도착")
         val pendingItem = (viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content)
             .messages.single()
@@ -414,6 +553,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         viewModel.sendText("중복 방지")
         val requestId = (viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content)
             .messages.single().requestId!!
@@ -431,12 +571,13 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         viewModel.sendText("늦은 ACK")
         val pendingItem = (viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content)
             .messages.single()
         val requestId = pendingItem.requestId!!
 
-        reconnectScheduler.advanceTimeBy(10L, TimeUnit.SECONDS)
+        reconnectScheduler.advanceTimeBy(15L, TimeUnit.SECONDS)
         shadowOf(Looper.getMainLooper()).idle()
         socketFactory.emitAck(requestId, message(messageId = 62L, mine = true, textMessage = "늦은 ACK"))
 
@@ -451,12 +592,13 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         viewModel.sendText("재시도 ACK")
         val firstState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         val failedItem = firstState.messages.single()
         val firstRequestId = failedItem.requestId!!
 
-        reconnectScheduler.advanceTimeBy(10L, TimeUnit.SECONDS)
+        reconnectScheduler.advanceTimeBy(15L, TimeUnit.SECONDS)
         shadowOf(Looper.getMainLooper()).idle()
         viewModel.retry(failedItem.localId!!)
         val retryRequestId = (viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content)
@@ -535,6 +677,36 @@ class DmChatRoomViewModelTest {
         assertEquals(listOf(1L, 2L), state.messages.map { it.messageId })
     }
 
+    @Test
+    fun `realtime 연결 중 상태는 JOINED 수신 전까지만 true이다`() {
+        api.enqueueOpenSuccess(openResponse(roomId = 10L))
+        viewModel.enter(roomId = 10L, creatorId = 0L)
+
+        viewModel.connectRealtime()
+
+        assertEquals(true, viewModel.isRealtimeConnectingLiveData.requireValue())
+
+        socketFactory.emitJoined()
+
+        assertEquals(false, viewModel.isRealtimeConnectingLiveData.requireValue())
+    }
+
+    @Test
+    fun `realtime 연결 실패와 leave는 연결 중 상태를 해제한다`() {
+        api.enqueueOpenSuccess(openResponse(roomId = 10L))
+        viewModel.enter(roomId = 10L, creatorId = 0L)
+
+        viewModel.connectRealtime()
+        socketFactory.emitFailure(IllegalStateException("network"))
+
+        assertEquals(false, viewModel.isRealtimeConnectingLiveData.requireValue())
+
+        viewModel.connectRealtime()
+        viewModel.leaveRealtime()
+
+        assertEquals(false, viewModel.isRealtimeConnectingLiveData.requireValue())
+    }
+
     @Test
     fun `realtime 연결 중 중복 connect 요청은 무시한다`() {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
@@ -587,6 +759,7 @@ class DmChatRoomViewModelTest {
         viewModel.enter(roomId = 10L, creatorId = 0L)
 
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         socketFactory.emitMessage(message(messageId = 3L, textMessage = "실시간"))
 
         val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
@@ -599,6 +772,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
         val beforeCallbackSize = viewModel.compositeDisposable.size()
 
         socketFactory.emitMessage(message(messageId = 3L, textMessage = "실시간1"))
@@ -725,6 +899,7 @@ class DmChatRoomViewModelTest {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
+        socketFactory.emitJoined()
 
         viewModel.leaveRealtime()
         viewModel.leaveRealtime()
@@ -773,6 +948,48 @@ class DmChatRoomViewModelTest {
         assertEquals("JOIN_ROOM", socketFactory.webSocket.sentTexts.lastJson().get("type").asString)
     }
 
+    @Test
+    fun `socket close callback은 foreground 상태에서 reconnect를 예약한다`() {
+        api.enqueueOpenSuccess(openResponse(roomId = 10L))
+        api.enqueueMessagesSuccess(messagesPage(messages = emptyList()))
+        viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
+        socketFactory.emitJoined()
+
+        socketFactory.emitClosed()
+        reconnectScheduler.advanceTimeBy(2999L, TimeUnit.MILLISECONDS)
+
+        assertEquals(false, viewModel.isRealtimeConnectedForTest())
+        assertEquals(1, socketFactory.closeCount)
+        assertEquals(1, socketFactory.connectCalls.size)
+
+        reconnectScheduler.advanceTimeBy(1L, TimeUnit.MILLISECONDS)
+
+        assertEquals(2, socketFactory.connectCalls.size)
+        assertEquals("JOIN_ROOM", socketFactory.webSocket.sentTexts.lastJson().get("type").asString)
+    }
+
+    @Test
+    fun `heartbeat PING 전송 실패는 timeout 대기 없이 socket close 후 reconnect를 예약한다`() {
+        api.enqueueOpenSuccess(openResponse(roomId = 10L))
+        api.enqueueMessagesSuccess(messagesPage(messages = emptyList()))
+        viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
+        socketFactory.emitJoined()
+        socketFactory.webSocket.sendResult = false
+
+        reconnectScheduler.advanceTimeBy(30L, TimeUnit.SECONDS)
+        reconnectScheduler.advanceTimeBy(2999L, TimeUnit.MILLISECONDS)
+
+        assertEquals(false, viewModel.isRealtimeConnectedForTest())
+        assertEquals(1, socketFactory.closeCount)
+        assertEquals(1, socketFactory.connectCalls.size)
+
+        reconnectScheduler.advanceTimeBy(1L, TimeUnit.MILLISECONDS)
+
+        assertEquals(2, socketFactory.connectCalls.size)
+    }
+
     @Test
     fun `leave는 heartbeat timeout과 reconnect 예약을 취소한다`() {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
@@ -1031,8 +1248,47 @@ class FakeDmChatApi : DmChatApi {
         limit: Int
     ): Single<ApiResponse<DmChatMessagesPageResponse>> {
         messagesCalls.add(MessagesCall(authHeader, roomId, cursor, limit))
-        return messagesResponses.removeFirst()
+        return if (messagesResponses.isEmpty()) {
+            Single.just(
+                ApiResponse(
+                    success = true,
+                    data = DmChatMessagesPageResponse(
+                        messages = emptyList(),
+                        hasMore = false,
+                        nextCursor = null
+                    )
+                )
+            )
+        } else {
+            messagesResponses.removeFirst()
         }
+    }
+
+    override fun sendDmVoiceMessage(
+        authHeader: String,
+        roomId: Long,
+        voiceMessageFile: MultipartBody.Part,
+        request: RequestBody
+    ): Single<ApiResponse<SendDmVoiceMessageResponse>> = Single.just(
+        ApiResponse(
+            success = true,
+            data = SendDmVoiceMessageResponse(
+                message = DmChatMessageResponse(
+                    messageId = 1L,
+                    messageType = "VOICE",
+                    mine = true,
+                    createdAt = 1000L,
+                    textMessage = null,
+                    voiceMessageUrl = "https://example.com/voice.m4a",
+                    senderId = 1L,
+                    senderNickname = "나",
+                    senderProfileImageUrl = "https://example.com/me.png"
+                ),
+                deliveredRealtime = true,
+                pushSent = false
+            )
+        )
+    )
 }
 
 class FakeWebSocketFactory {
@@ -1050,44 +1306,83 @@ class FakeWebSocketFactory {
     }
 
     fun emitJoined() {
-        webSocketListener?.onMessage(webSocket, "{\"type\":\"JOINED\",\"payload\":{}}")
+        webSocketListener?.onMessage(
+            webSocket,
+            "{\"type\":\"JOINED\",\"requestId\":\"join-request\",\"roomId\":10,\"payload\":{}}"
+        )
     }
 
     fun emitMessage(message: DmChatMessageResponse) {
         val json = Gson().toJson(message)
-        webSocketListener?.onMessage(webSocket, "{\"type\":\"MESSAGE\",\"payload\":{\"message\":$json}}")
+        webSocketListener?.onMessage(webSocket, "{\"type\":\"MESSAGE\",\"requestId\":null,\"roomId\":10,\"payload\":$json}")
     }
 
     fun emitMessage(requestId: String, message: DmChatMessageResponse) {
+        emitMessage(roomId = 10L, requestId = requestId, message = message)
+    }
+
+    fun emitMessage(roomId: Long?, message: DmChatMessageResponse) {
         val json = Gson().toJson(message)
         webSocketListener?.onMessage(
             webSocket,
-            "{\"type\":\"MESSAGE\",\"payload\":{\"requestId\":\"$requestId\",\"message\":$json}}"
+            "{\"type\":\"MESSAGE\",\"requestId\":null,\"roomId\":$roomId,\"payload\":$json}"
+        )
+    }
+
+    fun emitMessage(roomId: Long, requestId: String, message: DmChatMessageResponse) {
+        val json = Gson().toJson(message)
+        webSocketListener?.onMessage(
+            webSocket,
+            "{\"type\":\"MESSAGE\",\"requestId\":\"$requestId\",\"roomId\":$roomId,\"payload\":$json}"
         )
     }
 
     fun emitAck(requestId: String, message: DmChatMessageResponse) {
+        emitAck(roomId = 10L, requestId = requestId, message = message)
+    }
+
+    fun emitAck(roomId: Long?, requestId: String, message: DmChatMessageResponse) {
         val json = Gson().toJson(message)
         webSocketListener?.onMessage(
             webSocket,
-            "{\"type\":\"SEND_ACK\",\"payload\":{\"requestId\":\"$requestId\",\"message\":$json}}"
+            "{\"type\":\"SEND_ACK\",\"requestId\":\"$requestId\",\"roomId\":$roomId,\"payload\":$json}"
         )
     }
 
     fun emitError(requestId: String) {
+        val json = """
+            {"type":"ERROR","requestId":"$requestId","roomId":10,"payload":{"messageKey":"chat.message.send_failed"}}
+        """.trimIndent()
         webSocketListener?.onMessage(
             webSocket,
-            "{\"type\":\"ERROR\",\"payload\":{\"requestId\":\"$requestId\",\"code\":\"SEND_FAILED\",\"message\":\"failed\"}}"
+            json
+        )
+    }
+
+    fun emitJoinError(roomId: Long?) {
+        val json = """
+            {"type":"ERROR","requestId":"join-request","roomId":$roomId,"payload":{"messageKey":"chat.room.invalid_access"}}
+        """.trimIndent()
+        webSocketListener?.onMessage(
+            webSocket,
+            json
         )
     }
 
     fun emitPong() {
-        webSocketListener?.onMessage(webSocket, "{\"type\":\"PONG\",\"payload\":{}}")
+        webSocketListener?.onMessage(
+            webSocket,
+            "{\"type\":\"PONG\",\"requestId\":\"ping-request\",\"roomId\":10,\"payload\":{}}"
+        )
     }
 
     fun emitFailure(throwable: Throwable) {
         webSocketListener?.onFailure(webSocket, throwable, null)
     }
+
+    fun emitClosed() {
+        webSocketListener?.onClosed(webSocket, 1001, "going away")
+    }
 }
 
 class FakeWebSocket : WebSocket {