refactor(dm): 채팅 화면 전송을 WebSocket으로 전환한다

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

@@ -52,7 +52,7 @@ class DmChatRoomActivity : BaseActivity<ActivityDmChatRoomBinding>(
 
     override fun onStop() {
         isStarted = false
-        viewModel.disconnectRealtime()
+        viewModel.leaveRealtime()
         super.onStop()
     }
 

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

@@ -13,12 +13,12 @@ import kr.co.vividnext.sodalive.base.BaseViewModel
 import kr.co.vividnext.sodalive.common.ApiResponse
 import kr.co.vividnext.sodalive.common.SharedPreferenceManager
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.CreateDmChatRoomResponse
-import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatEventClient
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatMessageResponse
 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.SendDmChatMessageResponse
+import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatSocketClient
+import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatSocketEvent
 import kr.co.vividnext.sodalive.v2.main.chat.dm.model.DmChatMessageStatus
 import kr.co.vividnext.sodalive.v2.main.chat.dm.model.DmChatMessageUiItem
 import kr.co.vividnext.sodalive.v2.main.chat.dm.model.DmChatRoomUiState
@@ -46,7 +46,6 @@ class DmChatRoomViewModel(
     private var shouldReconnectRealtime: Boolean = false
     private var currentAuthToken: String = ""
     private var currentRealtimeToken: String = ""
-    private var isDisconnecting: Boolean = false
     private var reconnectDisposable: Disposable? = null
     private var localMessageSequence: Long = 0L
     private val mainHandler = Handler(Looper.getMainLooper())
@@ -187,16 +186,11 @@ class DmChatRoomViewModel(
         shouldReconnectRealtime = true
         reconnectDisposable?.dispose()
         reconnectDisposable = null
-        repository.connectRealtime(
+        repository.connectSocket(
             token = token,
-            roomId = roomId,
+            listener = object : DmChatSocketClient.Listener {
-            listener = object : DmChatEventClient.Listener {
+                override fun onEvent(event: DmChatSocketEvent) {
-                override fun onConnected() {
+                    scheduleRealtimeCallback { handleSocketEvent(event, token) }
-                    scheduleRealtimeCallback { syncLatestMessagesAfterReconnect(token = token) }
-                }
-
-                override fun onMessage(message: DmChatMessageResponse) {
-                    scheduleRealtimeCallback { onRealtimeMessage(message) }
                 }
 
                 override fun onFailure(throwable: Throwable) {
@@ -208,9 +202,10 @@ class DmChatRoomViewModel(
                 }
             }
         )
+        repository.sendJoinRoom(roomId)
     }
 
-    fun disconnectRealtime() {
+    fun leaveRealtime() {
         val roomId = currentRoomId
         if (roomId <= 0L) return
 
@@ -219,22 +214,8 @@ class DmChatRoomViewModel(
         isRealtimeConnected = false
         reconnectDisposable?.dispose()
         reconnectDisposable = null
-        repository.cancelRealtime()
+        repository.sendLeaveRoom(roomId)
-        if (isDisconnecting) return
+        repository.closeSocket()
-
-        isDisconnecting = true
-        compositeDisposable.add(
-            repository.disconnectRealtime(token = authToken(), roomId = roomId)
-                .subscribeOn(Schedulers.io())
-                .observeOn(AndroidSchedulers.mainThread())
-                .subscribe(
-                    { isDisconnecting = false },
-                    {
-                        isDisconnecting = false
-                        it.message?.let { message -> Logger.e(message) }
-                    }
-                )
-        )
     }
 
     private fun scheduleRealtimeReconnect() {
@@ -266,7 +247,7 @@ class DmChatRoomViewModel(
         mainHandler.removeCallbacksAndMessages(null)
         reconnectDisposable?.dispose()
         reconnectDisposable = null
-        repository.cancelRealtime()
+        repository.closeSocket()
         super.onCleared()
     }
 
@@ -302,22 +283,12 @@ class DmChatRoomViewModel(
     }
 
     private fun sendLocalMessage(localId: String, text: String) {
-        compositeDisposable.add(
+        val sent = repository.sendSocketText(
-            repository.sendTextMessage(
+            roomId = currentRoomId,
-                token = authToken(),
+            requestId = localId,
-                roomId = currentRoomId,
+            textMessage = text
-                textMessage = text
-            )
-                .subscribeOn(Schedulers.io())
-                .observeOn(AndroidSchedulers.mainThread())
-                .subscribe(
-                    { handleSendResult(localId, it) },
-                    {
-                        it.message?.let { message -> Logger.e(message) }
-                        markLocalMessageFailed(localId)
-                    }
-                )
         )
+        if (!sent) markLocalMessageFailed(localId)
     }
 
     private fun handleOpenRoomResult(response: ApiResponse<DmChatRoomOpenResponse>) {
@@ -356,12 +327,18 @@ class DmChatRoomViewModel(
         emitContent()
     }
 
-    private fun handleSendResult(
+    private fun handleSocketEvent(event: DmChatSocketEvent, token: String) {
-        localId: String,
+        when (event) {
-        response: ApiResponse<SendDmChatMessageResponse>
+            DmChatSocketEvent.Joined -> syncLatestMessagesAfterReconnect(token = token)
-    ) {
+            is DmChatSocketEvent.Message -> onRealtimeMessage(event.message)
-        val message = response.data?.message
+            is DmChatSocketEvent.SendAck -> handleSendAck(event.requestId, event.message)
-        val sentItem = if (response.success && message != null) message.toUiItem() else null
+            is DmChatSocketEvent.Error -> event.requestId?.let { markLocalMessageFailed(it) }
+            DmChatSocketEvent.Pong -> Unit
+        }
+    }
+
+    private fun handleSendAck(localId: String, message: DmChatMessageResponse) {
+        val sentItem = message.toUiItem()
         if (sentItem == null) {
             markLocalMessageFailed(localId)
             return

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

@@ -36,7 +36,7 @@ class DmChatRoomActivitySourceTest {
         assertTrue(source.contains("roomOpenedEventLiveData.observe(this)"))
         assertTrue(source.contains("if (it.consume() == true) connectRealtimeIfStarted()"))
         assertTrue(source.contains("connectRealtimeIfStarted()"))
-        assertTrue(source.contains("viewModel.disconnectRealtime()"))
+        assertTrue(source.contains("viewModel.leaveRealtime()"))
         assertFalse(source.contains("if (isStarted) viewModel.connectRealtime()"))
         assertTrue(!source.contains("character_type_badge"))
         assertTrue(!source.contains("notice_container"))

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

@@ -6,13 +6,14 @@ import android.os.Looper
 import androidx.lifecycle.LiveData
 import androidx.lifecycle.Observer
 import androidx.test.core.app.ApplicationProvider
+import com.google.gson.Gson
+import com.google.gson.JsonParser
 import io.reactivex.rxjava3.android.plugins.RxAndroidPlugins
 import io.reactivex.rxjava3.core.Scheduler
 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 io.reactivex.rxjava3.subjects.SingleSubject
 import kr.co.vividnext.sodalive.common.ApiResponse
 import kr.co.vividnext.sodalive.common.SharedPreferenceManager
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.CreateDmChatRoomRequest
@@ -20,14 +21,16 @@ import kr.co.vividnext.sodalive.v2.main.chat.dm.data.CreateDmChatRoomResponse
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatApi
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatMessageResponse
 import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatMessagesPageResponse
-import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatEventClient
-import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatRealtimeClient
 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.SendDmChatMessageResponse
+import kr.co.vividnext.sodalive.v2.main.chat.dm.data.DmChatSocketClient
-import kr.co.vividnext.sodalive.v2.main.chat.dm.data.SendDmTextMessageRequest
 import kr.co.vividnext.sodalive.v2.main.chat.dm.model.DmChatMessageStatus
 import kr.co.vividnext.sodalive.v2.main.chat.dm.model.DmChatRoomUiState
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.WebSocket
+import okhttp3.WebSocketListener
+import okio.ByteString
 import org.junit.After
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue
@@ -47,7 +50,8 @@ class DmChatRoomViewModelTest {
 
     private val context: Context = ApplicationProvider.getApplicationContext()
     private lateinit var api: FakeDmChatApi
-    private lateinit var realtimeClient: FakeDmChatRealtimeClient
+    private lateinit var socketFactory: FakeWebSocketFactory
+    private lateinit var socketClient: DmChatSocketClient
     private lateinit var reconnectScheduler: TestScheduler
     private lateinit var viewModel: DmChatRoomViewModel
 
@@ -58,10 +62,16 @@ class DmChatRoomViewModelTest {
         SharedPreferenceManager.init(context)
         SharedPreferenceManager.token = "test-token"
         api = FakeDmChatApi()
-        realtimeClient = FakeDmChatRealtimeClient()
+        socketFactory = FakeWebSocketFactory()
+        socketClient = DmChatSocketClient(
+            okHttpClient = OkHttpClient(),
+            gson = Gson(),
+            baseUrl = "https://api.example.com",
+            webSocketFactory = socketFactory::newWebSocket
+        )
         reconnectScheduler = TestScheduler()
         viewModel = DmChatRoomViewModel(
-            repository = DmChatRepository(api, realtimeClient),
+            repository = DmChatRepository(api, socketClient),
             reconnectScheduler = reconnectScheduler,
             tokenProvider = { "test-token" }
         )
@@ -148,35 +158,27 @@ class DmChatRoomViewModelTest {
 
         viewModel.sendText("   ")
 
-        assertTrue(api.sendCalls.isEmpty())
+        assertTrue(socketFactory.webSocket.sentTexts.isEmpty())
         val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         assertTrue(state.messages.isEmpty())
     }
 
     @Test
     fun `전송 직후 pending을 추가하고 성공 시 서버 메시지로 교체한다`() {
-        val pendingSend = SingleSubject.create<ApiResponse<SendDmChatMessageResponse>>()
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
-        api.enqueueSend(pendingSend)
         viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
 
         viewModel.sendText(" 안녕 ")
         val sendingState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
+        val localId = sendingState.messages.single().localId!!
 
-        assertEquals(listOf(SendCall("Bearer test-token", 10L, SendDmTextMessageRequest("안녕"))), api.sendCalls)
+        assertEquals("SEND_TEXT", socketFactory.webSocket.sentJsonAt(1).get("type").asString)
+        assertEquals(localId, socketFactory.webSocket.sentJsonAt(1).getAsJsonObject("payload").get("requestId").asString)
         assertEquals(DmChatMessageStatus.SENDING, sendingState.messages.single().status)
         assertEquals("안녕", sendingState.messages.single().textMessage)
 
-        pendingSend.onSuccess(
+        socketFactory.emitAck(localId, message(messageId = 30L, mine = true, textMessage = "안녕"))
-            ApiResponse(
-                success = true,
-                data = SendDmChatMessageResponse(
-                    message = message(messageId = 30L, mine = true, textMessage = "안녕"),
-                    deliveredRealtime = true,
-                    pushSent = false
-                )
-            )
-        )
         val sentState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         assertEquals(listOf(30L), sentState.messages.map { it.messageId })
         assertEquals(DmChatMessageStatus.SENT, sentState.messages.single().status)
@@ -184,15 +186,14 @@ class DmChatRoomViewModelTest {
 
     @Test
     fun `전송 중 새 전송 중복 요청은 무시한다`() {
-        val pendingSend = SingleSubject.create<ApiResponse<SendDmChatMessageResponse>>()
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
-        api.enqueueSend(pendingSend)
         viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
 
         viewModel.sendText("안녕")
         viewModel.sendText("안녕")
 
-        assertEquals(1, api.sendCalls.size)
+        assertEquals(2, socketFactory.webSocket.sentTexts.size)
         val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         assertEquals(1, state.messages.size)
     }
@@ -200,16 +201,18 @@ class DmChatRoomViewModelTest {
     @Test
     fun `전송 실패는 pending 메시지를 FAILED로 바꾸고 retry 성공 시 교체한다`() {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
-        api.enqueueSend(Single.error(IllegalStateException("network")))
         viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
+        socketFactory.webSocket.sendResult = false
 
         viewModel.sendText("안녕")
         val failedState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         val failedItem = failedState.messages.single()
         assertEquals(DmChatMessageStatus.FAILED, failedItem.status)
 
-        api.enqueueSendSuccess(message(messageId = 40L, mine = true, textMessage = "안녕"))
+        socketFactory.webSocket.sendResult = true
         viewModel.retry(failedItem.localId!!)
+        socketFactory.emitAck(failedItem.localId, message(messageId = 40L, mine = true, textMessage = "안녕"))
 
         val retriedState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         assertEquals(listOf(40L), retriedState.messages.map { it.messageId })
@@ -218,27 +221,18 @@ class DmChatRoomViewModelTest {
 
     @Test
     fun `retry 중 SSE echo가 먼저 와도 성공 교체 후 messageId 중복을 남기지 않는다`() {
-        val pendingRetry = SingleSubject.create<ApiResponse<SendDmChatMessageResponse>>()
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
-        api.enqueueSend(Single.error(IllegalStateException("network")))
         viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
+        socketFactory.webSocket.sendResult = false
         viewModel.sendText("안녕")
         val failedState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         val failedItem = failedState.messages.single()
-        api.enqueueSend(pendingRetry)
 
+        socketFactory.webSocket.sendResult = true
         viewModel.retry(failedItem.localId!!)
         viewModel.onRealtimeMessage(message(messageId = 45L, mine = true, textMessage = "안녕"))
-        pendingRetry.onSuccess(
+        socketFactory.emitAck(failedItem.localId, message(messageId = 45L, mine = true, textMessage = "안녕"))
-            ApiResponse(
-                success = true,
-                data = SendDmChatMessageResponse(
-                    message = message(messageId = 45L, mine = true, textMessage = "안녕"),
-                    deliveredRealtime = true,
-                    pushSent = false
-                )
-            )
-        )
 
         val retriedState = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         assertEquals(listOf(45L), retriedState.messages.map { it.messageId })
@@ -260,23 +254,14 @@ class DmChatRoomViewModelTest {
 
     @Test
     fun `SSE echo가 전송 성공보다 먼저 와도 성공 교체 후 messageId 중복을 남기지 않는다`() {
-        val pendingSend = SingleSubject.create<ApiResponse<SendDmChatMessageResponse>>()
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
-        api.enqueueSend(pendingSend)
         viewModel.enter(roomId = 10L, creatorId = 0L)
+        viewModel.connectRealtime()
 
         viewModel.sendText("안녕")
+        val localId = (viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content).messages.single().localId!!
         viewModel.onRealtimeMessage(message(messageId = 50L, mine = true, textMessage = "안녕"))
-        pendingSend.onSuccess(
+        socketFactory.emitAck(localId, message(messageId = 50L, mine = true, textMessage = "안녕"))
-            ApiResponse(
-                success = true,
-                data = SendDmChatMessageResponse(
-                    message = message(messageId = 50L, mine = true, textMessage = "안녕"),
-                    deliveredRealtime = true,
-                    pushSent = false
-                )
-            )
-        )
 
         val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         assertEquals(listOf(50L), state.messages.map { it.messageId })
@@ -316,11 +301,10 @@ class DmChatRoomViewModelTest {
     @Test
     fun `roomId가 없으면 realtime 연결과 disconnect를 요청하지 않는다`() {
         viewModel.connectRealtime()
-        viewModel.disconnectRealtime()
+        viewModel.leaveRealtime()
 
-        assertTrue(realtimeClient.connectCalls.isEmpty())
+        assertTrue(socketFactory.connectCalls.isEmpty())
-        assertEquals(0, realtimeClient.cancelCalls)
+        assertEquals(0, socketFactory.closeCount)
-        assertTrue(api.disconnectCalls.isEmpty())
     }
 
     @Test
@@ -339,9 +323,9 @@ class DmChatRoomViewModelTest {
         viewModel.enter(roomId = 10L, creatorId = 0L)
 
         viewModel.connectRealtime()
-        realtimeClient.listener?.onConnected()
+        socketFactory.emitJoined()
 
-        assertEquals(listOf(RealtimeConnectCall("test-token", 10L)), realtimeClient.connectCalls)
+        assertEquals(listOf(RealtimeConnectCall("test-token", 10L)), socketFactory.connectCalls)
         assertEquals(listOf(MessagesCall("Bearer test-token", 10L, null, 20)), api.messagesCalls)
         val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         assertEquals(listOf(1L, 2L), state.messages.map { it.messageId })
@@ -355,7 +339,7 @@ class DmChatRoomViewModelTest {
         viewModel.connectRealtime()
         viewModel.connectRealtime()
 
-        assertEquals(listOf(RealtimeConnectCall("test-token", 10L)), realtimeClient.connectCalls)
+        assertEquals(listOf(RealtimeConnectCall("test-token", 10L)), socketFactory.connectCalls)
     }
 
     @Test
@@ -380,20 +364,15 @@ class DmChatRoomViewModelTest {
 
     @Test
     fun `disconnect 진행 중 빠른 reconnect 시 crash 없이 connect를 허용한다`() {
-        val pendingDisconnect = SingleSubject.create<ApiResponse<Boolean>>()
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
-        api.enqueueDisconnect(pendingDisconnect)
         viewModel.enter(roomId = 10L, creatorId = 0L)
 
         viewModel.connectRealtime()
-        viewModel.disconnectRealtime()
+        viewModel.leaveRealtime()
         viewModel.connectRealtime()
 
-        assertEquals(2, realtimeClient.connectCalls.size)
+        assertEquals(2, socketFactory.connectCalls.size)
-        assertEquals(1, realtimeClient.cancelCalls)
+        assertTrue(socketFactory.closeCount >= 1)
-        assertEquals(listOf(DisconnectCall("Bearer test-token", 10L)), api.disconnectCalls)
-
-        pendingDisconnect.onSuccess(ApiResponse(success = true, data = true))
     }
 
     @Test
@@ -402,7 +381,7 @@ class DmChatRoomViewModelTest {
         viewModel.enter(roomId = 10L, creatorId = 0L)
 
         viewModel.connectRealtime()
-        realtimeClient.listener?.onMessage(message(messageId = 3L, textMessage = "실시간"))
+        socketFactory.emitMessage(message(messageId = 3L, textMessage = "실시간"))
 
         val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         assertEquals(listOf(3L), state.messages.map { it.messageId })
@@ -416,9 +395,9 @@ class DmChatRoomViewModelTest {
         viewModel.connectRealtime()
         val beforeCallbackSize = viewModel.compositeDisposable.size()
 
-        realtimeClient.listener?.onMessage(message(messageId = 3L, textMessage = "실시간1"))
+        socketFactory.emitMessage(message(messageId = 3L, textMessage = "실시간1"))
-        realtimeClient.listener?.onMessage(message(messageId = 4L, textMessage = "실시간2"))
+        socketFactory.emitMessage(message(messageId = 4L, textMessage = "실시간2"))
-        realtimeClient.listener?.onMessage(message(messageId = 5L, textMessage = "실시간3"))
+        socketFactory.emitMessage(message(messageId = 5L, textMessage = "실시간3"))
 
         assertEquals(beforeCallbackSize, viewModel.compositeDisposable.size())
         val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
@@ -434,8 +413,7 @@ class DmChatRoomViewModelTest {
         assertTrue(source.contains("scheduleRealtimeCallback"))
         assertTrue(source.contains("Looper.myLooper() == Looper.getMainLooper()"))
         assertTrue(source.contains("mainHandler.post { action() }"))
-        assertTrue(source.contains("scheduleRealtimeCallback { syncLatestMessagesAfterReconnect(token = token) }"))
+        assertTrue(source.contains("scheduleRealtimeCallback { handleSocketEvent(event, token) }"))
-        assertTrue(source.contains("scheduleRealtimeCallback { onRealtimeMessage(message) }"))
     }
 
     @Test
@@ -447,16 +425,16 @@ class DmChatRoomViewModelTest {
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
 
-        realtimeClient.listener?.onFailure(IllegalStateException("network"))
+        socketFactory.emitFailure(IllegalStateException("network"))
         reconnectScheduler.advanceTimeBy(2999L, TimeUnit.MILLISECONDS)
 
-        assertEquals(1, realtimeClient.connectCalls.size)
+        assertEquals(1, socketFactory.connectCalls.size)
 
         reconnectScheduler.advanceTimeBy(1L, TimeUnit.MILLISECONDS)
-        assertEquals(2, realtimeClient.connectCalls.size)
+        assertEquals(2, socketFactory.connectCalls.size)
-        assertEquals(RealtimeConnectCall("test-token", 10L), realtimeClient.connectCalls[1])
+        assertEquals(RealtimeConnectCall("test-token", 10L), socketFactory.connectCalls[1])
 
-        realtimeClient.listener?.onConnected()
+        socketFactory.emitJoined()
         assertEquals(listOf(MessagesCall("Bearer test-token", 10L, null, 20)), api.messagesCalls)
         val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         assertEquals(listOf(1L, 2L), state.messages.map { it.messageId })
@@ -479,94 +457,79 @@ class DmChatRoomViewModelTest {
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
 
-        realtimeClient.listener?.onFailure(IllegalStateException("network-1"))
+        socketFactory.emitFailure(IllegalStateException("network-1"))
         reconnectScheduler.advanceTimeBy(3L, TimeUnit.SECONDS)
-        realtimeClient.listener?.onFailure(IllegalStateException("network-2"))
+        socketFactory.emitFailure(IllegalStateException("network-2"))
         reconnectScheduler.advanceTimeBy(2999L, TimeUnit.MILLISECONDS)
 
-        assertEquals(2, realtimeClient.connectCalls.size)
+        assertEquals(2, socketFactory.connectCalls.size)
 
         reconnectScheduler.advanceTimeBy(1L, TimeUnit.MILLISECONDS)
 
-        assertEquals(3, realtimeClient.connectCalls.size)
+        assertEquals(3, socketFactory.connectCalls.size)
     }
 
     @Test
-    fun `disconnect는 예약된 SSE 재연결을 취소한다`() {
+    fun `leave는 예약된 SSE 재연결을 취소한다`() {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
 
-        realtimeClient.listener?.onFailure(IllegalStateException("network"))
+        socketFactory.emitFailure(IllegalStateException("network"))
-        viewModel.disconnectRealtime()
+        viewModel.leaveRealtime()
         reconnectScheduler.advanceTimeBy(3L, TimeUnit.SECONDS)
 
-        assertEquals(1, realtimeClient.connectCalls.size)
+        assertEquals(1, socketFactory.connectCalls.size)
-        assertEquals(1, realtimeClient.cancelCalls)
+        assertTrue(socketFactory.closeCount >= 1)
-        assertEquals(listOf(DisconnectCall("Bearer test-token", 10L)), api.disconnectCalls)
     }
 
     @Test
-    fun `예약 재연결 실행 후 main callback 전 disconnect되면 새 SSE 연결을 만들지 않는다`() {
+    fun `예약 재연결 실행 후 main callback 전 leave되면 새 SSE 연결을 만들지 않는다`() {
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
 
-        realtimeClient.listener?.onFailure(IllegalStateException("network"))
+        socketFactory.emitFailure(IllegalStateException("network"))
-        viewModel.disconnectRealtime()
+        viewModel.leaveRealtime()
         reconnectScheduler.advanceTimeBy(3L, TimeUnit.SECONDS)
 
-        assertEquals(listOf(RealtimeConnectCall("test-token", 10L)), realtimeClient.connectCalls)
+        assertEquals(listOf(RealtimeConnectCall("test-token", 10L)), socketFactory.connectCalls)
-        assertEquals(1, realtimeClient.cancelCalls)
+        assertTrue(socketFactory.closeCount >= 1)
-        assertEquals(listOf(DisconnectCall("Bearer test-token", 10L)), api.disconnectCalls)
     }
 
     @Test
-    fun `realtime disconnect 중 중복 요청은 무시하고 완료 후 다시 요청할 수 있다`() {
+    fun `realtime leave 중 중복 요청은 close를 반복할 수 있다`() {
-        val pendingDisconnect = SingleSubject.create<ApiResponse<Boolean>>()
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
-        api.enqueueDisconnect(pendingDisconnect)
-        api.enqueueDisconnectSuccess()
         viewModel.enter(roomId = 10L, creatorId = 0L)
 
-        viewModel.disconnectRealtime()
+        viewModel.leaveRealtime()
-        viewModel.disconnectRealtime()
+        viewModel.leaveRealtime()
 
-        assertEquals(2, realtimeClient.cancelCalls)
+        assertEquals(0, socketFactory.closeCount)
-        assertEquals(listOf(DisconnectCall("Bearer test-token", 10L)), api.disconnectCalls)
+        viewModel.leaveRealtime()
 
-        pendingDisconnect.onSuccess(ApiResponse(success = true, data = true))
+        assertEquals(0, socketFactory.closeCount)
-        viewModel.disconnectRealtime()
-
-        assertEquals(3, realtimeClient.cancelCalls)
-        assertEquals(2, api.disconnectCalls.size)
     }
 
     @Test
-    fun `disconnect API 진행 중 다시 background로 가면 새 SSE 연결도 cancel하고 API 중복 호출은 하지 않는다`() {
+    fun `leave 후 다시 background로 가면 새 소켓도 close한다`() {
-        val pendingDisconnect = SingleSubject.create<ApiResponse<Boolean>>()
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
-        api.enqueueDisconnect(pendingDisconnect)
         viewModel.enter(roomId = 10L, creatorId = 0L)
 
         viewModel.connectRealtime()
-        viewModel.disconnectRealtime()
+        viewModel.leaveRealtime()
         viewModel.connectRealtime()
-        viewModel.disconnectRealtime()
+        viewModel.leaveRealtime()
 
-        assertEquals(2, realtimeClient.cancelCalls)
+        assertEquals(2, socketFactory.connectCalls.size)
-        assertEquals(listOf(DisconnectCall("Bearer test-token", 10L)), api.disconnectCalls)
-
-        pendingDisconnect.onSuccess(ApiResponse(success = true, data = true))
     }
 
     @Test
-    fun `realtime disconnect 실패는 채팅 상태를 Error로 바꾸지 않는다`() {
+    fun `realtime leave는 채팅 상태를 Error로 바꾸지 않는다`() {
         api.enqueueOpenSuccess(openResponse(roomId = 10L, messages = listOf(message(messageId = 1L, textMessage = "기존"))))
-        api.enqueueDisconnect(Single.error(IllegalStateException("network")))
         viewModel.enter(roomId = 10L, creatorId = 0L)
 
-        viewModel.disconnectRealtime()
+        viewModel.leaveRealtime()
 
         val state = viewModel.chatRoomStateLiveData.requireValue() as DmChatRoomUiState.Content
         assertEquals(listOf(1L), state.messages.map { it.messageId })
@@ -574,18 +537,15 @@ class DmChatRoomViewModelTest {
 
     @Test
     fun `onCleared는 realtime 연결과 disposable을 정리한다`() {
-        val pendingDisconnect = SingleSubject.create<ApiResponse<Boolean>>()
         api.enqueueOpenSuccess(openResponse(roomId = 10L))
-        api.enqueueDisconnect(pendingDisconnect)
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
-        viewModel.disconnectRealtime()
+        viewModel.leaveRealtime()
 
         viewModel.invokeOnCleared()
 
-        assertEquals(2, realtimeClient.cancelCalls)
+        assertEquals(1, socketFactory.closeCount)
         assertTrue(viewModel.compositeDisposable.isDisposed)
-        assertTrue(pendingDisconnect.hasObservers().not())
     }
 
     @Test
@@ -594,12 +554,12 @@ class DmChatRoomViewModelTest {
         viewModel.enter(roomId = 10L, creatorId = 0L)
         viewModel.connectRealtime()
 
-        realtimeClient.listener?.onFailure(IllegalStateException("network"))
+        socketFactory.emitFailure(IllegalStateException("network"))
         viewModel.invokeOnCleared()
         reconnectScheduler.advanceTimeBy(3L, TimeUnit.SECONDS)
 
-        assertEquals(listOf(RealtimeConnectCall("test-token", 10L)), realtimeClient.connectCalls)
+        assertEquals(listOf(RealtimeConnectCall("test-token", 10L)), socketFactory.connectCalls)
-        assertEquals(1, realtimeClient.cancelCalls)
+        assertEquals(1, socketFactory.closeCount)
     }
 
     @Test
@@ -610,7 +570,7 @@ class DmChatRoomViewModelTest {
         viewModel.connectRealtime()
 
         Thread {
-            realtimeClient.listener?.onMessage(message(messageId = 99L, textMessage = "제거 대상"))
+            socketFactory.emitMessage(message(messageId = 99L, textMessage = "제거 대상"))
             postedLatch.countDown()
         }.start()
         assertTrue(postedLatch.await(2, TimeUnit.SECONDS))
@@ -712,17 +672,6 @@ data class MessagesCall(
     val limit: Int
 )
 
-data class SendCall(
-    val authHeader: String,
-    val roomId: Long,
-    val request: SendDmTextMessageRequest
-)
-
-data class DisconnectCall(
-    val authHeader: String,
-    val roomId: Long
-)
-
 data class RealtimeConnectCall(
     val token: String,
     val roomId: Long
@@ -732,14 +681,10 @@ class FakeDmChatApi : DmChatApi {
     val createCalls = mutableListOf<CreateCall>()
     val openCalls = mutableListOf<OpenCall>()
     val messagesCalls = mutableListOf<MessagesCall>()
-    val sendCalls = mutableListOf<SendCall>()
-    val disconnectCalls = mutableListOf<DisconnectCall>()
 
     private val createResponses = ArrayDeque<Single<ApiResponse<CreateDmChatRoomResponse>>>()
     private val openResponses = ArrayDeque<Single<ApiResponse<DmChatRoomOpenResponse>>>()
     private val messagesResponses = ArrayDeque<Single<ApiResponse<DmChatMessagesPageResponse>>>()
-    private val sendResponses = ArrayDeque<Single<ApiResponse<SendDmChatMessageResponse>>>()
-    private val disconnectResponses = ArrayDeque<Single<ApiResponse<Boolean>>>()
 
     fun enqueueCreateSuccess(response: CreateDmChatRoomResponse) {
         createResponses.addLast(Single.just(ApiResponse(success = true, data = response)))
@@ -757,33 +702,6 @@ class FakeDmChatApi : DmChatApi {
         messagesResponses.addLast(response)
     }
 
-    fun enqueueSend(response: Single<ApiResponse<SendDmChatMessageResponse>>) {
-        sendResponses.addLast(response)
-    }
-
-    fun enqueueDisconnect(response: Single<ApiResponse<Boolean>>) {
-        disconnectResponses.addLast(response)
-    }
-
-    fun enqueueDisconnectSuccess() {
-        disconnectResponses.addLast(Single.just(ApiResponse(success = true, data = true)))
-    }
-
-    fun enqueueSendSuccess(message: DmChatMessageResponse) {
-        sendResponses.addLast(
-            Single.just(
-                ApiResponse(
-                    success = true,
-                    data = SendDmChatMessageResponse(
-                        message = message,
-                        deliveredRealtime = true,
-                        pushSent = false
-                    )
-                )
-            )
-        )
-    }
-
     override fun createDmChatRoom(
         authHeader: String,
         request: CreateDmChatRoomRequest
@@ -810,41 +728,61 @@ class FakeDmChatApi : DmChatApi {
         messagesCalls.add(MessagesCall(authHeader, roomId, cursor, limit))
         return messagesResponses.removeFirst()
     }
-
-    override fun sendDmTextMessage(
-        authHeader: String,
-        roomId: Long,
-        request: SendDmTextMessageRequest
-    ): Single<ApiResponse<SendDmChatMessageResponse>> {
-        sendCalls.add(SendCall(authHeader, roomId, request))
-        return sendResponses.removeFirst()
-    }
-
-    override fun disconnectRealtime(
-        authHeader: String,
-        roomId: Long
-    ): Single<ApiResponse<Boolean>> {
-        disconnectCalls.add(DisconnectCall(authHeader, roomId))
-        return disconnectResponses.removeFirstOrNull() ?: Single.just(ApiResponse(success = true, data = true))
-    }
 }
 
-class FakeDmChatRealtimeClient : DmChatRealtimeClient {
+class FakeWebSocketFactory {
+    val webSocket = FakeWebSocket()
     val connectCalls = mutableListOf<RealtimeConnectCall>()
-    var cancelCalls = 0
+    var webSocketListener: WebSocketListener? = null
-    var listener: DmChatEventClient.Listener? = null
+    val closeCount: Int
+        get() = webSocket.closeCount
 
-    override fun connect(
+    fun newWebSocket(request: Request, listener: WebSocketListener): WebSocket {
-        token: String,
+        val token = request.header("Authorization")?.removePrefix("Bearer ").orEmpty()
-        roomId: Long,
+        connectCalls.add(RealtimeConnectCall(token = token, roomId = 10L))
-        listener: DmChatEventClient.Listener
+        webSocketListener = listener
-    ) {
+        return webSocket
-        connectCalls.add(RealtimeConnectCall(token, roomId))
-        this.listener = listener
     }
 
-    override fun cancel() {
+    fun emitJoined() {
-        cancelCalls += 1
+        webSocketListener?.onMessage(webSocket, "{\"type\":\"JOINED\",\"payload\":{}}")
-        listener = null
+    }
+
+    fun emitMessage(message: DmChatMessageResponse) {
+        val json = Gson().toJson(message)
+        webSocketListener?.onMessage(webSocket, "{\"type\":\"MESSAGE\",\"payload\":{\"message\":$json}}")
+    }
+
+    fun emitAck(requestId: String, message: DmChatMessageResponse) {
+        val json = Gson().toJson(message)
+        webSocketListener?.onMessage(
+            webSocket,
+            "{\"type\":\"SEND_ACK\",\"payload\":{\"requestId\":\"$requestId\",\"message\":$json}}"
+        )
+    }
+
+    fun emitFailure(throwable: Throwable) {
+        webSocketListener?.onFailure(webSocket, throwable, null)
     }
 }
+
+class FakeWebSocket : WebSocket {
+    val sentTexts = mutableListOf<String>()
+    var sendResult = true
+    var closeCount = 0
+
+    override fun request(): Request = Request.Builder().url("wss://example.com").build()
+    override fun queueSize(): Long = 0L
+    override fun send(text: String): Boolean {
+        sentTexts += text
+        return sendResult
+    }
+    override fun send(bytes: ByteString): Boolean = sendResult
+    override fun close(code: Int, reason: String?): Boolean {
+        closeCount += 1
+        return true
+    }
+    override fun cancel() = Unit
+
+    fun sentJsonAt(index: Int) = JsonParser.parseString(sentTexts[index]).asJsonObject
+}