From 871f4e73e8e0aa1d060cbd3fa89255e66b30729d Mon Sep 17 00:00:00 2001
From: klaus <klaus@vividnext.co.kr>
Date: Thu, 11 Jun 2026 11:16:19 +0900
Subject: [PATCH] =?UTF-8?q?feat(chat):=20DM=20SSE=20=EC=9E=AC=EC=97=B0?=
 =?UTF-8?q?=EA=B2=B0=20=EA=B8=B0=EB=B0=98=EC=9D=84=20=EC=B6=94=EA=B0=80?=
 =?UTF-8?q?=ED=95=9C=EB=8B=A4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../v2/main/chat/dm/data/DmChatEventClient.kt | 28 +++++++++++++++----
 .../v2/main/chat/dm/data/DmChatRepository.kt  | 17 ++++++++++-
 .../v2/main/chat/dm/DmChatEventClientTest.kt  | 25 +++++++++++++++++
 3 files changed, 64 insertions(+), 6 deletions(-)

diff --git a/app/src/main/java/kr/co/vividnext/sodalive/v2/main/chat/dm/data/DmChatEventClient.kt b/app/src/main/java/kr/co/vividnext/sodalive/v2/main/chat/dm/data/DmChatEventClient.kt
index dbfd78e0..a4b2af22 100644
--- a/app/src/main/java/kr/co/vividnext/sodalive/v2/main/chat/dm/data/DmChatEventClient.kt
+++ b/app/src/main/java/kr/co/vividnext/sodalive/v2/main/chat/dm/data/DmChatEventClient.kt
@@ -49,11 +49,21 @@ class DmChatEventParser(private val gson: Gson) {
     }
 }
 
+interface DmChatRealtimeClient {
+    fun connect(
+        token: String,
+        roomId: Long,
+        listener: DmChatEventClient.Listener
+    )
+
+    fun cancel()
+}
+
 class DmChatEventClient(
     private val okHttpClient: OkHttpClient,
     gson: Gson,
     private val baseUrl: String
-) {
+) : DmChatRealtimeClient {
     interface Listener {
         fun onConnected()
         fun onMessage(message: DmChatMessageResponse)
@@ -67,7 +77,7 @@ class DmChatEventClient(
     private var listener: Listener? = null
 
     @Synchronized
-    fun connect(
+    override fun connect(
         token: String,
         roomId: Long,
         listener: Listener
@@ -105,7 +115,7 @@ class DmChatEventClient(
     }
 
     @Synchronized
-    fun cancel() {
+    override fun cancel() {
         call?.cancel()
         call = null
         listener = null
@@ -115,7 +125,12 @@ class DmChatEventClient(
         reader.use { bufferedReader ->
             val frame = StringBuilder()
             while (!call.isCanceled()) {
-                val line = bufferedReader.readLine() ?: break
+                val line = bufferedReader.readLine()
+                if (line == null) {
+                    if (frame.isNotEmpty()) dispatch(frame.toString())
+                    notifyStreamClosed(call)
+                    return
+                }
                 if (line.isBlank()) {
                     dispatch(frame.toString())
                     frame.clear()
@@ -123,10 +138,13 @@ class DmChatEventClient(
                     frame.append(line).append('\n')
                 }
             }
-            if (frame.isNotEmpty()) dispatch(frame.toString())
         }
     }
 
+    private fun notifyStreamClosed(call: Call) {
+        if (!call.isCanceled()) listener?.onFailure(IOException("SSE stream closed"))
+    }
+
     private fun dispatch(frame: String) {
         when (val event = parser.parse(frame)) {
             DmChatEventParser.Event.Connected -> listener?.onConnected()
diff --git a/app/src/main/java/kr/co/vividnext/sodalive/v2/main/chat/dm/data/DmChatRepository.kt b/app/src/main/java/kr/co/vividnext/sodalive/v2/main/chat/dm/data/DmChatRepository.kt
index 4fe09ded..e4174e85 100644
--- a/app/src/main/java/kr/co/vividnext/sodalive/v2/main/chat/dm/data/DmChatRepository.kt
+++ b/app/src/main/java/kr/co/vividnext/sodalive/v2/main/chat/dm/data/DmChatRepository.kt
@@ -3,7 +3,10 @@ package kr.co.vividnext.sodalive.v2.main.chat.dm.data
 import io.reactivex.rxjava3.core.Single
 import kr.co.vividnext.sodalive.common.ApiResponse
 
-class DmChatRepository(private val api: DmChatApi) {
+class DmChatRepository(
+    private val api: DmChatApi,
+    private val realtimeClient: DmChatRealtimeClient? = null
+) {
     fun createOrGetRoom(
         token: String,
         creatorId: Long
@@ -52,6 +55,18 @@ class DmChatRepository(private val api: DmChatApi) {
         roomId = roomId
     )
 
+    fun connectRealtime(
+        token: String,
+        roomId: Long,
+        listener: DmChatEventClient.Listener
+    ) {
+        realtimeClient?.connect(token = token, roomId = roomId, listener = listener)
+    }
+
+    fun cancelRealtime() {
+        realtimeClient?.cancel()
+    }
+
     private fun bearer(token: String) = "Bearer $token"
 
     private companion object {
diff --git a/app/src/test/java/kr/co/vividnext/sodalive/v2/main/chat/dm/DmChatEventClientTest.kt b/app/src/test/java/kr/co/vividnext/sodalive/v2/main/chat/dm/DmChatEventClientTest.kt
index a491802c..c1c29093 100644
--- a/app/src/test/java/kr/co/vividnext/sodalive/v2/main/chat/dm/DmChatEventClientTest.kt
+++ b/app/src/test/java/kr/co/vividnext/sodalive/v2/main/chat/dm/DmChatEventClientTest.kt
@@ -66,6 +66,31 @@ class DmChatEventClientTest {
         assertEquals("안녕하세요", receivedMessage?.textMessage)
     }
 
+    @Test
+    fun `취소되지 않은 SSE stream이 EOF로 종료되면 failure callback으로 전달된다`() {
+        val failureLatch = CountDownLatch(1)
+        var failure: Throwable? = null
+        val client = clientWithResponse(
+            code = 200,
+            body = "event: connected\n\n"
+        )
+
+        client.connect(
+            token = "test-token",
+            roomId = 10L,
+            listener = object : TestListener() {
+                override fun onFailure(throwable: Throwable) {
+                    failure = throwable
+                    failureLatch.countDown()
+                }
+            }
+        )
+
+        failureLatch.await(2, TimeUnit.SECONDS)
+        assertNotNull(failure)
+        assertEquals("SSE stream closed", failure?.message)
+    }
+
     private fun clientWithResponse(
         code: Int,
         body: String