Introduce event dispatcher and fix heartbeat, auth reply events

bl-client/src/commonMain/kotlin/cn/rtast/bldm/client/bldm.kt

@@ -4,88 +4,108 @@
  * Date: 2026/8/30
  */
 
-
-@file:OptIn(InternalBldmApi::class)
-
 package cn.rtast.bldm.client
 
-import cn.rtast.bldm.codec.BLDMConstants
-import cn.rtast.bldm.codec.annotations.InternalBldmApi
-import cn.rtast.bldm.codec.data.DMServerConf
-import cn.rtast.bldm.codec.data.RealRoomId
-import cn.rtast.bldm.codec.data.UserNavData
-import cn.rtast.bldm.codec.protocol.Packet
+import cn.rtast.bldm.client.util.fromJson
+import cn.rtast.bldm.client.util.sendPacket
+import cn.rtast.bldm.codec.event.DanmuEventDispatcher
 import cn.rtast.bldm.codec.protocol.codec.PacketDecoder
 import cn.rtast.bldm.codec.protocol.codec.PacketEncoder
-import cn.rtast.bldm.codec.protocol.event.PacketEvents
-import cn.rtast.bldm.codec.util._fromJson
 import cn.rtast.bldm.codec.util.parseBuvidAndUid
+import cn.rtast.bldm.codec.util.signWbi
+import cn.rtast.bldm.dto.BldmConstant
+import cn.rtast.bldm.dto.DMServerConf
+import cn.rtast.bldm.dto.RealRoomId
+import cn.rtast.bldm.dto.UserNavData
 import io.ktor.client.plugins.websocket.*
 import io.ktor.client.request.*
 import io.ktor.client.statement.*
 import io.ktor.http.*
 import io.ktor.websocket.*
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.isActive
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.*
 import kotlin.time.Duration.Companion.seconds
 
-private val HEARTBEAT_INTERVAL = 30.seconds
+public class BldmClient internal constructor(
+    private val roomId: Long,
+    private val cookie: String? = null,
+) : DanmuEventDispatcher(), AutoCloseable {
+    private companion object {
+        private val HEARTBEAT_INTERVAL = 30.seconds
+    }
 
-public suspend fun connectToBlDM(
-    roomId: Long,
-    cookie: String?,
-    events: PacketEvents.() -> Unit,
-) {
-    val parsedCookie = parseBuvidAndUid(cookie ?: "")
-    val packetEncoder = PacketEncoder()
-    val packetDecoder = PacketDecoder(events)
-    val realRoomId = httpClient.get(BLDMConstants.REAL_ROOM_ID_URL + roomId)
-        .bodyAsText()._fromJson<RealRoomId>()
-    val wbi = httpClient.get(BLDMConstants.USER_NAV_URL)
-        .bodyAsText()._fromJson<UserNavData>().signWbi(realRoomId.data.roomId)
-    val serverConf = httpClient.get(BLDMConstants.DM_SERVER_CONF_URL + "?$wbi") {
-        cookie?.let { header(HttpHeaders.Cookie, it) }
-    }.bodyAsText()._fromJson<DMServerConf>().data
-    val wsUrl = "ws://${serverConf.hostList.first().host}:${serverConf.hostList.first().wsPort}/sub"
+    private val packetEncoder = PacketEncoder()
+    private val decoder = PacketDecoder(this)
+    private val clientScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
+    private var connectionJob: Job? = null
+    private var session: DefaultClientWebSocketSession? = null
 
-    httpClient.webSocket(wsUrl, {
-//        header(HttpHeaders.Host, "${serverConf.hostList.first().host}:${serverConf.hostList.first().wssPort}")
-//        header(HttpHeaders.Origin, "https://www.bilibili.com")
-    }) {
-        val authPacket = packetEncoder.authPacket(
-            parsedCookie.second ?: 0,
-            realRoomId.data.roomId,
-            parsedCookie.first ?: "",
-            serverConf.token
-        )
-        sendPacket(authPacket)
-        val heartbeatJob = launch {
-            val heartbeatPacket = packetEncoder.heartbeatPacket()
-            while (isActive) {
-                delay(HEARTBEAT_INTERVAL)
-                runCatching {
-                    sendPacket(heartbeatPacket)
-                }.onFailure {
-                    println("heartbeat failed ${it.message}")
+    private suspend fun internalConnect() {
+        val parsedCookie = parseBuvidAndUid(cookie ?: "")
+        val realRoomId = httpClient.get(BldmConstant.REAL_ROOM_ID_URL + "?id=$roomId")
+            .bodyAsText().fromJson<RealRoomId>()
+        val navData = httpClient.get(BldmConstant.USER_NAV_URL)
+            .bodyAsText().fromJson<UserNavData>().data.wbiImg
+        val wbi = signWbi(realRoomId.data.roomId, navData.imgUrl, navData.subUrl)
+        val serverConf = httpClient.get(BldmConstant.DM_SERVER_CONF_URL + "?$wbi") {
+            cookie?.let { header(HttpHeaders.Cookie, it) }
+        }.bodyAsText().fromJson<DMServerConf>().data
+
+        httpClient.webSocket(serverConf.hostList.first().tlsWsAddress) {
+            session = this
+            val authPacket = packetEncoder.authPacket(
+                parsedCookie.second ?: 0,
+                realRoomId.data.roomId,
+                parsedCookie.first ?: "",
+                serverConf.token
+            )
+            sendPacket(authPacket)
+            val heartbeatJob = launch {
+                val heartbeatPacket = packetEncoder.heartbeatPacket()
+                while (isActive) {
+                    delay(HEARTBEAT_INTERVAL)
+                    runCatching {
+                        sendPacket(heartbeatPacket)
+                    }.onFailure {
+                    }
                 }
             }
-        }
-        try {
-            for (frame in incoming) {
-                if (frame !is Frame.Binary) continue
-                val data = frame.readBytes()
-                packetDecoder.decode(data)
+
+            try {
+                for (frame in incoming) {
+                    if (frame !is Frame.Binary) continue
+                    val data = frame.readBytes()
+                    decoder.decode(data)
+                }
+            } catch (e: Exception) {
+                if (session?.isActive == true) e.printStackTrace()
+            } finally {
+                heartbeatJob.cancel()
+                session = null
             }
-        } catch (e: Exception) {
-            e.printStackTrace()
-        } finally {
-            heartbeatJob.cancel()
-            println("Closed")
         }
     }
-    httpClient.close()
+
+    public fun connect() {
+        connectionJob = clientScope.launch { internalConnect() }
+    }
+
+    public fun connectBlocking(): Unit = runBlocking { internalConnect() }
+
+    public suspend fun disconnect(reason: CloseReason = CloseReason(CloseReason.Codes.NORMAL, "")) {
+        session?.close(reason)
+        session = null
+        connectionJob?.cancel()
+    }
+
+    public fun disconnectBlocking(reason: CloseReason = CloseReason(CloseReason.Codes.NORMAL, "")): Unit =
+        runBlocking { disconnect(reason) }
+
+    override fun close() {
+        runBlocking { disconnect() }
+        clientScope.cancel()
+    }
 }
 
-internal suspend fun DefaultWebSocketSession.sendPacket(packet: Packet) =
-    send(Frame.Binary(true, packet.toByteArray()))
+@Suppress("FunctionName")
+public fun BLDMClient(roomId: Long, cookie: String? = null): BldmClient =
+    BldmClient(roomId, cookie)

bl-client/src/commonMain/kotlin/cn/rtast/bldm/client/util/_json.kt

@@ -0,0 +1,20 @@
+/*
+ * Copyright © 2026 RTAkland
+ * Author: RTAkland
+ * Date: 2026/8/31
+ */
+
+@file:Suppress("ObjectPropertyName", "FunctionName")
+
+package cn.rtast.bldm.client.util
+
+import kotlinx.serialization.json.Json
+
+internal val _json: Json = Json {
+    ignoreUnknownKeys = true
+    isLenient = true
+}
+
+internal inline fun <reified T> String.fromJson(): T = _json.decodeFromString(this)
+
+internal inline fun <reified T> T.encodeJson(): String = _json.encodeToString(this)

bl-client/src/commonMain/kotlin/cn/rtast/bldm/client/util/ws_binary_send.kt

@@ -0,0 +1,15 @@
+/*
+ * Copyright © 2026 RTAkland
+ * Author: RTAkland
+ * Date: 2026/8/31
+ */
+
+
+package cn.rtast.bldm.client.util
+
+import cn.rtast.bldm.codec.protocol.Packet
+import io.ktor.websocket.DefaultWebSocketSession
+import io.ktor.websocket.Frame
+
+internal suspend fun DefaultWebSocketSession.sendPacket(packet: Packet) =
+    send(Frame.Binary(true, packet.toByteArray()))

bl-client/src/commonTest/kotlin/test/TestClient.kt

@@ -7,13 +7,16 @@
 
 package test
 
-import cn.rtast.bldm.client.connectToBlDM
+import cn.rtast.bldm.client.BLDMClient
+import cn.rtast.bldm.codec.event.DanmuEvent
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.test.runTest
 import kotlinx.io.buffered
 import kotlinx.io.files.Path
 import kotlinx.io.files.SystemFileSystem
 import kotlinx.io.readString
 import kotlin.test.Test
+import kotlin.time.Duration.Companion.seconds
 
 class TestClient {
 
@@ -22,18 +25,15 @@ class TestClient {
 
     @Test
     fun `test bldm client`() = runTest {
-        connectToBlDM(6, cookie) {
-            onHeartbeat {
-                println("Heartbeat")
-            }
+        val client = BLDMClient(7777, cookie)
+        client.on<DanmuEvent.HeartbeatEvent> { println("Heartbeat") }
+        client.on<DanmuEvent.AuthReplyEvent> { println("AuthReply") }
+        client.on<DanmuEvent.MessageEvent> { println(it.content) }
+        client.connectBlocking()
 
-            onMessage {
-                println(it.content)
-            }
-
-            onAuthReply {
-                println("auth reply")
-            }
-        }
+        // remove lines below if you use client.connectBlocking()
+//        while (true) {
+//            delay(1.seconds)
+//        }
     }
 }