feat: add event connection state transition api

Aliorpse · Aliorpse · commit a7f7bc3c4a9c · 2026-02-21T06:59:14.000+08:00
diff --git a/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/core/SaltifyApplication.kt b/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/core/SaltifyApplication.kt
@@ -12,16 +12,20 @@ import io.ktor.serialization.kotlinx.*
 import io.ktor.serialization.kotlinx.json.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asSharedFlow
+import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.serialization.json.decodeFromJsonElement
 import org.ntqqrev.milky.*
 import org.ntqqrev.saltify.annotation.WithApiExtension
-import org.ntqqrev.saltify.dsl.SaltifyConfig
+import org.ntqqrev.saltify.dsl.SaltifyApplicationConfig
 import org.ntqqrev.saltify.dsl.SaltifyPluginContext
+import org.ntqqrev.saltify.exception.ApiCallException
+import org.ntqqrev.saltify.model.EventConnectionState
 import org.ntqqrev.saltify.model.EventConnectionType
 import org.ntqqrev.saltify.model.SaltifyComponentType
-import org.ntqqrev.saltify.exception.ApiCallException
 import org.ntqqrev.saltify.util.coroutine.SaltifyComponent
 import org.ntqqrev.saltify.util.coroutine.SaltifyExceptionHandlerProvider
 import kotlin.coroutines.CoroutineContext
@@ -30,14 +34,14 @@ import kotlin.coroutines.CoroutineContext
  * 一个 Saltify 应用实例
  */
 @WithApiExtension
-public sealed class SaltifyApplication(private val config: SaltifyConfig) : AutoCloseable {
+public sealed class SaltifyApplication(protected val config: SaltifyApplicationConfig) : AutoCloseable {
     public companion object {
         /**
          * 创建一个 Saltify 应用实例。
          */
-        public operator fun invoke(block: SaltifyConfig.() -> Unit): SaltifyApplication {
-            val config = SaltifyConfig().apply(block)
-            return when (config.eventConnectionType) {
+        public operator fun invoke(block: SaltifyApplicationConfig.() -> Unit): SaltifyApplication {
+            val config = SaltifyApplicationConfig().apply(block)
+            return when (config.eventConnectionConfig.type) {
                 EventConnectionType.WebSocket -> SaltifyApplicationWebSocket(config)
                 EventConnectionType.SSE -> SaltifyApplicationSSE(config)
             }
@@ -61,20 +65,19 @@ public sealed class SaltifyApplication(private val config: SaltifyConfig) : Auto
     public val exceptionFlow: SharedFlow<Pair<CoroutineContext, Throwable>> =
         exceptionHandlerProvider.exceptionFlow.asSharedFlow()
 
+    protected val eventConnectionState: MutableStateFlow<EventConnectionState> =
+        MutableStateFlow(EventConnectionState.Disconnected(null))
+
+    /**
+     * 事件服务连接状态流。
+     */
+    public val eventConnectionStateFlow: StateFlow<EventConnectionState> = eventConnectionState.asStateFlow()
+
     internal val applicationScope: CoroutineScope = CoroutineScope(
         SaltifyComponent(SaltifyComponentType.Application, "SaltifyApplication") +
             exceptionHandlerProvider.handler
     )
 
-    @PublishedApi
-    internal val extensionScope: CoroutineScope = CoroutineScope(
-        applicationScope.coroutineContext +
-            SupervisorJob(applicationScope.coroutineContext.job) +
-            SaltifyComponent(SaltifyComponentType.Extension, "SaltifyExtension")
-    )
-
-    protected val addressBaseNormalized: String = config.addressBase.trimEnd('/')
-
     protected val events: MutableSharedFlow<Event> = MutableSharedFlow(extraBufferCapacity = 64)
 
     /**
@@ -84,7 +87,16 @@ public sealed class SaltifyApplication(private val config: SaltifyConfig) : Auto
      */
     public val eventFlow: SharedFlow<Event> = events.asSharedFlow()
 
-    private val activePlugins = mutableListOf<SaltifyPluginContext>()
+    @PublishedApi
+    internal val extensionScope: CoroutineScope = CoroutineScope(
+        applicationScope.coroutineContext +
+            SupervisorJob(applicationScope.coroutineContext.job) +
+            SaltifyComponent(SaltifyComponentType.Extension, "SaltifyExtension")
+    )
+
+    protected val addressBaseNormalized: String = config.addressBase.trimEnd('/')
+
+    private val loadedPlugins = mutableListOf<SaltifyPluginContext>()
 
     @PublishedApi
     internal val httpClient: HttpClient = HttpClient {
@@ -97,7 +109,7 @@ public sealed class SaltifyApplication(private val config: SaltifyConfig) : Auto
             config.accessToken?.let { header(HttpHeaders.Authorization, "Bearer $it") }
         }
 
-        when (config.eventConnectionType) {
+        when (config.eventConnectionConfig.type) {
             EventConnectionType.WebSocket -> install(WebSockets.Plugin) {
                 contentConverter = KotlinxWebsocketSerializationConverter(milkyJsonModule)
             }
@@ -116,8 +128,11 @@ public sealed class SaltifyApplication(private val config: SaltifyConfig) : Auto
 
             val context = SaltifyPluginContext(this, pluginScope)
             plugin.setup(context)
+            loadedPlugins.add(context)
 
-            activePlugins.add(context)
+            pluginScope.launch {
+                context.onStartHooks.forEach { it() }
+            }
         }
     }
 
@@ -147,26 +162,17 @@ public sealed class SaltifyApplication(private val config: SaltifyConfig) : Auto
     /**
      * 连接事件服务。需要在监听事件时调用。请搭配 [disconnectEvent] 使用。
      */
-    public open suspend fun connectEvent(): Unit = coroutineScope {
-        activePlugins.map { context ->
-            context.launch {
-                context.onStartHooks.forEach { it() }
-            }
-        }.joinAll()
-    }
+    public abstract suspend fun connectEvent()
 
     /**
      * 断开事件服务。请搭配 [connectEvent] 使用。
      */
-    public open suspend fun disconnectEvent(): Unit = coroutineScope {
-        activePlugins.map { context ->
-            context.launch {
-                context.onStopHooks.forEach { it() }
-            }
-        }.joinAll()
-    }
+    public abstract suspend fun disconnectEvent()
 
     override fun close() {
+        loadedPlugins.forEach {
+            it.onStopHooks.forEach { block -> block() }
+        }
         httpClient.close()
         applicationScope.cancel()
     }
diff --git a/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/core/SaltifyApplicationSSE.kt b/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/core/SaltifyApplicationSSE.kt
@@ -6,32 +6,53 @@ import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.launch
 import org.ntqqrev.milky.Event
 import org.ntqqrev.milky.milkyJsonModule
-import org.ntqqrev.saltify.dsl.SaltifyConfig
+import org.ntqqrev.saltify.dsl.SaltifyApplicationConfig
+import org.ntqqrev.saltify.model.EventConnectionState
+import org.ntqqrev.saltify.model.EventConnectionType
+import org.ntqqrev.saltify.util.coroutine.withRetry
 
-public class SaltifyApplicationSSE(config: SaltifyConfig) : SaltifyApplication(config) {
+public class SaltifyApplicationSSE(config: SaltifyApplicationConfig) : SaltifyApplication(config) {
     private var connectionJob: Job? = null
 
     override suspend fun connectEvent() {
         connectionJob?.cancelAndJoin()
+        eventConnectionState.emit(EventConnectionState.Connecting)
 
         connectionJob = applicationScope.launch {
-            httpClient.sse("$addressBaseNormalized/event") {
-                incoming.collect { sseEvent ->
-                    if (sseEvent.event == "milky_event") {
-                        sseEvent.data?.let { data ->
-                            val event = milkyJsonModule.decodeFromString<Event>(data)
-                            events.emit(event)
+            withRetry(
+                config.eventConnectionConfig.maxReconnectionAttempts,
+                config.eventConnectionConfig.baseReconnectionInterval,
+                config.eventConnectionConfig.maxReconnectionInterval,
+                config.eventConnectionConfig.autoReconnect,
+                onRetry = { throwable, retryCount ->
+                    eventConnectionState.emit(EventConnectionState.Reconnecting(throwable, retryCount))
+                },
+                onFailure = {
+                    eventConnectionState.emit(EventConnectionState.Disconnected(it))
+                },
+                block = {
+                    eventConnectionState.emit(
+                        EventConnectionState.Connected(
+                            EventConnectionType.SSE, this@SaltifyApplicationSSE
+                        )
+                    )
+
+                    httpClient.sse("$addressBaseNormalized/event") {
+                        incoming.collect { sseEvent ->
+                            if (sseEvent.event == "milky_event") {
+                                sseEvent.data?.let { data ->
+                                    val event = milkyJsonModule.decodeFromString<Event>(data)
+                                    events.emit(event)
+                                }
+                            }
                         }
                     }
                 }
-            }
+            )
         }
-
-        super.connectEvent()
     }
 
     override suspend fun disconnectEvent() {
         connectionJob?.cancelAndJoin()
-        super.disconnectEvent()
     }
 }
diff --git a/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/core/SaltifyApplicationWebSocket.kt b/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/core/SaltifyApplicationWebSocket.kt
@@ -6,30 +6,51 @@ import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import org.ntqqrev.milky.Event
-import org.ntqqrev.saltify.dsl.SaltifyConfig
+import org.ntqqrev.saltify.dsl.SaltifyApplicationConfig
+import org.ntqqrev.saltify.model.EventConnectionState
+import org.ntqqrev.saltify.model.EventConnectionType
+import org.ntqqrev.saltify.util.coroutine.withRetry
 
-public class SaltifyApplicationWebSocket(config: SaltifyConfig) : SaltifyApplication(config) {
+public class SaltifyApplicationWebSocket(config: SaltifyApplicationConfig) : SaltifyApplication(config) {
     private var connectionJob: Job? = null
 
     override suspend fun connectEvent() {
         connectionJob?.cancelAndJoin()
+        eventConnectionState.emit(EventConnectionState.Connecting)
 
         connectionJob = applicationScope.launch {
             val urlString = "$addressBaseNormalized/event".replaceFirst("http", "ws")
 
-            httpClient.webSocket(urlString) {
-                while (isActive) {
-                    val event = receiveDeserialized<Event>()
-                    events.emit(event)
+            withRetry(
+                config.eventConnectionConfig.maxReconnectionAttempts,
+                config.eventConnectionConfig.baseReconnectionInterval,
+                config.eventConnectionConfig.maxReconnectionInterval,
+                config.eventConnectionConfig.autoReconnect,
+                onRetry = { throwable, retryCount ->
+                    eventConnectionState.emit(EventConnectionState.Reconnecting(throwable, retryCount))
+                },
+                onFailure = {
+                    eventConnectionState.emit(EventConnectionState.Disconnected(it))
+                },
+                block = {
+                    httpClient.webSocket(urlString) {
+                        eventConnectionState.emit(
+                            EventConnectionState.Connected(
+                                EventConnectionType.WebSocket, this@SaltifyApplicationWebSocket
+                            )
+                        )
+
+                        while (isActive) {
+                            val event = receiveDeserialized<Event>()
+                            events.emit(event)
+                        }
+                    }
                 }
-            }
+            )
         }
-
-        super.connectEvent()
     }
 
     override suspend fun disconnectEvent() {
         connectionJob?.cancelAndJoin()
-        super.disconnectEvent()
     }
 }
diff --git a/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/dsl/EventConnectionConfig.kt b/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/dsl/EventConnectionConfig.kt
@@ -0,0 +1,33 @@
+package org.ntqqrev.saltify.dsl
+
+import org.ntqqrev.saltify.annotation.SaltifyDsl
+import org.ntqqrev.saltify.model.EventConnectionType
+
+@Suppress("MagicNumber")
+@SaltifyDsl
+public class EventConnectionConfig {
+    /**
+     * 事件服务使用的协议。
+     */
+    public var type: EventConnectionType = EventConnectionType.WebSocket
+
+    /**
+     * 是否启用自动重连。
+     */
+    public var autoReconnect: Boolean = true
+
+    /**
+     * 最小自动重连间隔。
+     */
+    public var baseReconnectionInterval: Long = 500L
+
+    /**
+     * 最大自动重连间隔。
+     */
+    public var maxReconnectionInterval: Long = 10000L
+
+    /**
+     * 最大重试次数。
+     */
+    public var maxReconnectionAttempts: Int = 5
+}
diff --git a/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/dsl/SaltifyApplicationConfig.kt b/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/dsl/SaltifyApplicationConfig.kt
@@ -1,26 +1,29 @@
 package org.ntqqrev.saltify.dsl
 
 import org.ntqqrev.saltify.annotation.SaltifyDsl
-import org.ntqqrev.saltify.model.EventConnectionType
 
 @SaltifyDsl
-public class SaltifyConfig {
+public class SaltifyApplicationConfig {
     /**
      * 接口根地址，如：`https://localhost:3000`。
      */
     public var addressBase: String = ""
 
+    internal var eventConnectionConfig = EventConnectionConfig()
+
     /**
-     * 事件服务使用的协议。
+     * 事件服务配置。
      */
-    public var eventConnectionType: EventConnectionType = EventConnectionType.WebSocket
+    public fun eventConnection(block: EventConnectionConfig.() -> Unit) {
+        eventConnectionConfig.block()
+    }
 
     /**
      * 访问令牌，无需 Bearer。
      */
     public var accessToken: String? = null
-    internal val installedPlugins = mutableListOf<SaltifyPlugin>()
 
+    internal val installedPlugins = mutableListOf<SaltifyPlugin>()
     public fun install(plugin: SaltifyPlugin) {
         installedPlugins.add(plugin)
     }
diff --git a/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/dsl/SaltifyPluginContext.kt b/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/dsl/SaltifyPluginContext.kt
@@ -23,19 +23,19 @@ public class SaltifyPluginContext internal constructor(
     @PublishedApi internal val pluginScope: CoroutineScope
 ) : CoroutineScope by pluginScope {
     internal val onStartHooks = mutableListOf<suspend () -> Unit>()
-    internal val onStopHooks = mutableListOf<suspend () -> Unit>()
+    internal val onStopHooks = mutableListOf<() -> Unit>()
 
     /**
-     * 插件被加载，即 [SaltifyApplication.connectEvent] 后执行的逻辑。
+     * 插件被加载，即 [SaltifyApplication.Companion.invoke] 后执行的逻辑。
      */
     public fun onStart(block: suspend () -> Unit) {
         onStartHooks.add(block)
     }
 
     /**
-     * 插件被卸载，即 [SaltifyApplication.disconnectEvent] 后执行的逻辑。
+     * 插件被卸载，即 [SaltifyApplication.close] 前执行的逻辑。
      */
-    public fun onStop(block: suspend () -> Unit) {
+    public fun onStop(block: () -> Unit) {
         onStopHooks.add(block)
     }
 
diff --git a/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/model/EventConnectionState.kt b/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/model/EventConnectionState.kt
@@ -0,0 +1,16 @@
+package org.ntqqrev.saltify.model
+
+import org.ntqqrev.saltify.core.SaltifyApplication
+
+/**
+ * Saltify 应用的事件服务连接状态
+ */
+public sealed class EventConnectionState {
+    public data class Connected(
+        val type: EventConnectionType,
+        val instance: SaltifyApplication
+    ) : EventConnectionState()
+    public data class Disconnected(val throwable: Throwable?) : EventConnectionState()
+    public data class Reconnecting(val throwable: Throwable, val attempt: Int) : EventConnectionState()
+    public object Connecting : EventConnectionState()
+}
diff --git a/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/util/coroutine/WithRetry.kt b/saltify-core/src/commonMain/kotlin/org/ntqqrev/saltify/util/coroutine/WithRetry.kt
diff --git a/saltify-core/src/jvmTest/kotlin/org/ntqqrev/saltify/PluginTest.kt b/saltify-core/src/jvmTest/kotlin/org/ntqqrev/saltify/PluginTest.kt
diff --git a/saltify-docs/content/quick-tour.md b/saltify-docs/content/quick-tour.md
