From c643aca7682b727014c9d6cb5e88a7adb016a995 Mon Sep 17 00:00:00 2001
From: ForteScarlet <ForteScarlet@163.com>
Date: Fri, 5 Jan 2024 21:42:39 +0800
Subject: [PATCH] =?UTF-8?q?=E5=B0=9D=E8=AF=95=E4=BF=AE=E5=A4=8D/=E4=BC=98?=
 =?UTF-8?q?=E5=8C=96=E6=96=AD=E7=BA=BF=E9=87=8D=E8=BF=9E=E9=97=AE=E9=A2=98?=
 =?UTF-8?q?=20fix=20#116?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 buildSrc/src/main/kotlin/P.kt                 |  2 +-
 .../forte/simbot/kook/internal/BotStates.kt   | 49 +++++++++++++------
 2 files changed, 36 insertions(+), 15 deletions(-)

diff --git a/buildSrc/src/main/kotlin/P.kt b/buildSrc/src/main/kotlin/P.kt
index 407145e9..b10d06b7 100644
--- a/buildSrc/src/main/kotlin/P.kt
+++ b/buildSrc/src/main/kotlin/P.kt
@@ -68,7 +68,7 @@ object P : ProjectDetail() {
         0, 0
     )
 
-    private val alphaSuffix = v("beta2")
+    private val alphaSuffix = v("beta3")
 
     override val version: Version = baseVersion - alphaSuffix
 
diff --git a/simbot-component-kook-stdlib/src/commonMain/kotlin/love/forte/simbot/kook/internal/BotStates.kt b/simbot-component-kook-stdlib/src/commonMain/kotlin/love/forte/simbot/kook/internal/BotStates.kt
index b44c7f44..196d62f6 100644
--- a/simbot-component-kook-stdlib/src/commonMain/kotlin/love/forte/simbot/kook/internal/BotStates.kt
+++ b/simbot-component-kook-stdlib/src/commonMain/kotlin/love/forte/simbot/kook/internal/BotStates.kt
@@ -25,9 +25,7 @@ import kotlinx.atomicfu.AtomicLong
 import kotlinx.atomicfu.atomic
 import kotlinx.atomicfu.updateAndGet
 import kotlinx.coroutines.*
-import kotlinx.coroutines.channels.BufferOverflow
-import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.channels.ChannelResult
+import kotlinx.coroutines.channels.*
 import kotlinx.coroutines.flow.*
 import kotlinx.serialization.SerializationException
 import kotlinx.serialization.json.*
@@ -43,6 +41,7 @@ import love.forte.simbot.kook.event.UnknownExtra
 import love.forte.simbot.logger.Logger
 import kotlin.math.max
 import kotlin.time.Duration.Companion.seconds
+import kotlinx.coroutines.CancellationException as CreateCancellationException
 
 
 internal sealed class State : love.forte.simbot.util.stageloop.State<State>() {
@@ -394,7 +393,7 @@ private class CreateClient(
                 }
             }.catch { e ->
                 bot.eventLogger.error("Event process flow on error. Cause: {}", e.message, e)
-            }.launchIn(session)
+            }.launchIn(bot)
 
         return EventProcessJob(job, channel)
     }
@@ -457,19 +456,40 @@ private class Receiving(
 
         try {
             eventLogger.trace("Receiving next frame...")
-            val frame = session.incoming.receive()
+            val frameCatching = session.incoming.receiveCatching()
+            frameCatching.onFailure { cause ->
+                eventLogger.error("Receiving next frame on failure: {}, close current session and try to reconnect", cause?.message, cause)
+                val cancelException = CreateCancellationException("Receiving next frame on failure,, close current session and try to reconnect", cause)
+                session.cancel(cancelException)
+                return Reconnect(bot, botLogger, client.isCompress, client.sn.value, hello.d.sessionId)
+            }
+            frameCatching.onClosed { cause ->
+                eventLogger.error("Receiving next frame on closed: {}, close current session and try to reconnect", cause?.message, cause)
+                val cancelException = CreateCancellationException("Receiving next frame on failure,, close current session and try to reconnect", cause)
+                session.cancel(cancelException)
+                return Reconnect(bot, botLogger, client.isCompress, client.sn.value, hello.d.sessionId)
+            }
+
+            val frame = frameCatching.getOrThrow()
+//            val frame = session.incoming.receive()
             eventLogger.trace("Next frame: {}", frame)
 
-            when (frame) {
-                is Frame.Text -> return processSignalString(frame.readText())
-                is Frame.Binary -> return processSignalString(frame.readToText())
-                else -> {
-                    eventLogger.trace("Other frame: {}", frame)
+            try {
+                when (frame) {
+                    is Frame.Text -> return processSignalString(frame.readText())
+                    is Frame.Binary -> return processSignalString(frame.readToText())
+                    else -> {
+                        eventLogger.trace("Other frame: {}", frame)
+                    }
                 }
+            } catch (processException: Throwable) {
+                eventLogger.error("An exception was thrown while processing the event frame", processException)
             }
 
             return this
-        } catch (cancellation: CancellationException) {
+        } catch (cancellation: CreateCancellationException) {
+            // 似乎不会到这儿来？
+            session.cancel()
             return if (!bot.isActive) {
                 botLogger.warn("Session (and bot) is cancelled: {}", cancellation.message)
                 botLogger.debug("Session (and bot) is cancelled: {}", cancellation.message, cancellation)
@@ -482,9 +502,10 @@ private class Receiving(
             }
 
         } catch (e: Throwable) {
-            botLogger.error("Session received frame failed: {}, try to re-receive.", e.message, e)
-            // next: self
-            return this
+            botLogger.error("Session received frame failed: {}, cancel current session and try reconnect", e.message, e)
+            session.cancel(CreateCancellationException("unexpected exception was received", e))
+            // next: Reconnect
+            return Reconnect(bot, botLogger, client.isCompress, client.sn.value, hello.d.sessionId)
         }
     }