Exception Handling; Normal login tests

mirai-core/src/commonMain/kotlin/network/components/PacketCodec.kt

@@ -50,6 +50,18 @@ internal interface PacketCodec {
     }
 }
 
+internal class OicqDecodingException(
+    val targetException: Throwable
+) : RuntimeException(
+    null, targetException,
+    true, // enableSuppression
+    false, // writableStackTrace
+) {
+    override fun getStackTrace(): Array<StackTraceElement> {
+        return targetException.stackTrace
+    }
+}
+
 internal class PacketCodecImpl : PacketCodec {
 
     override fun decodeRaw(client: SsoSession, input: ByteReadPacket): RawIncomingPacket = input.run {
@@ -87,7 +99,13 @@ internal class PacketCodecImpl : PacketCodec {
                     2 -> RawIncomingPacket(
                         raw.commandName,
                         raw.sequenceId,
-                        raw.body.withUse { parseOicqResponse(client) }
+                        raw.body.withUse {
+                            try {
+                                parseOicqResponse(client)
+                            } catch (e: Throwable) {
+                                throw OicqDecodingException(e)
+                            }
+                        }
                     )
                     else -> error("Unknown flag2=$flag2")
                 }

mirai-core/src/commonMain/kotlin/network/impl/netty/NettyNetworkHandler.kt

@@ -31,6 +31,7 @@ import net.mamoe.mirai.internal.network.handler.logger
 import net.mamoe.mirai.internal.network.handler.state.StateObserver
 import net.mamoe.mirai.internal.network.protocol.packet.OutgoingPacket
 import net.mamoe.mirai.utils.*
+import java.io.EOFException
 import java.net.SocketAddress
 import kotlin.coroutines.CoroutineContext
 import io.netty.channel.Channel as NettyChannel
@@ -58,6 +59,29 @@ internal open class NettyNetworkHandler(
         return "NettyNetworkHandler(context=$context, address=$address)"
     }
 
+
+    ///////////////////////////////////////////////////////////////////////////
+    // exception handling
+    ///////////////////////////////////////////////////////////////////////////
+    protected open fun handleExceptionInDecoding(error: Throwable) {
+        if (error is OicqDecodingException) {
+            if (error.targetException is EOFException) return
+            throw error.targetException
+        }
+        throw error
+    }
+
+    protected open fun handlePipelineException(ctx: ChannelHandlerContext, error: Throwable) {
+        context.bot.logger.error(error)
+        synchronized(this) {
+            if (_state !is StateConnecting) {
+                setState { StateConnecting(ExceptionCollector(error)) }
+            } else {
+                close(error)
+            }
+        }
+    }
+
     ///////////////////////////////////////////////////////////////////////////
     // netty conn.
     ///////////////////////////////////////////////////////////////////////////
@@ -67,9 +91,13 @@ internal open class NettyNetworkHandler(
         private val ssoProcessor: SsoProcessor by lazy { context[SsoProcessor] }
 
         override fun channelRead0(ctx: ChannelHandlerContext, msg: ByteBuf) {
-            ctx.fireChannelRead(msg.toReadPacket().use { packet ->
-                packetCodec.decodeRaw(ssoProcessor.ssoSession, packet)
-            })
+            kotlin.runCatching {
+                ctx.fireChannelRead(msg.toReadPacket().use { packet ->
+                    packetCodec.decodeRaw(ssoProcessor.ssoSession, packet)
+                })
+            }.onFailure { error ->
+                handleExceptionInDecoding(error)
+            }
         }
     }
 
@@ -90,6 +118,11 @@ internal open class NettyNetworkHandler(
 
     protected open fun setupChannelPipeline(pipeline: ChannelPipeline, decodePipeline: PacketDecodePipeline) {
         pipeline
+            .addLast(object : ChannelInboundHandlerAdapter() {
+                override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
+                    handlePipelineException(ctx, cause)
+                }
+            })
             .addLast(OutgoingPacketEncoder())
             .addLast(LengthFieldBasedFrameDecoder(Int.MAX_VALUE, 0, 4, -4, 4))
             .addLast(ByteBufToIncomingPacketDecoder())

mirai-core/src/commonTest/kotlin/network/impl/netty/AbstractNettyNHTest.kt

@@ -11,6 +11,7 @@ package net.mamoe.mirai.internal.network.impl.netty
 
 import io.netty.channel.Channel
 import io.netty.channel.embedded.EmbeddedChannel
+import io.netty.util.ReferenceCountUtil
 import kotlinx.coroutines.CompletableDeferred
 import net.mamoe.mirai.internal.network.framework.AbstractRealNetworkHandlerTest
 import net.mamoe.mirai.internal.network.handler.NetworkHandlerContext
@@ -42,7 +43,30 @@ internal open class TestNettyNH(
 }
 
 internal abstract class AbstractNettyNHTest : AbstractRealNetworkHandlerTest<TestNettyNH>() {
-    val channel = EmbeddedChannel()
+    var fakeServer: (NettyNHTestChannel.(msg: Any?) -> Unit)? = null
+
+    internal inner class NettyNHTestChannel : EmbeddedChannel() {
+        public /*internal*/ override fun doRegister() {
+            super.doRegister() // Set channel state to ACTIVE
+            // Drop old handlers
+            pipeline().let { p ->
+                while (p.first() != null) {
+                    p.removeFirst()
+                }
+            }
+        }
+
+        override fun handleInboundMessage(msg: Any?) {
+            ReferenceCountUtil.release(msg) // Not handled, Drop
+        }
+
+        override fun handleOutboundMessage(msg: Any?) {
+            fakeServer?.invoke(this, msg) ?: ReferenceCountUtil.release(msg)
+        }
+    }
+
+    val channel = NettyNHTestChannel()
+
     override val network: TestNettyNH get() = bot.network as TestNettyNH
 
     override val factory: NetworkHandlerFactory<TestNettyNH> =
@@ -50,7 +74,10 @@ internal abstract class AbstractNettyNHTest : AbstractRealNetworkHandlerTest<Tes
             override fun create(context: NetworkHandlerContext, address: SocketAddress): TestNettyNH {
                 return object : TestNettyNH(context, address) {
                     override suspend fun createConnection(decodePipeline: PacketDecodePipeline): Channel =
-                        channel.apply { setupChannelPipeline(pipeline(), decodePipeline) }
+                        channel.apply {
+                            doRegister() // restart channel
+                            setupChannelPipeline(pipeline(), decodePipeline)
+                        }
                 }
             }
         }

mirai-core/src/commonTest/kotlin/network/impl/netty/NettyBotNormalLoginTest.kt

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2019-2021 Mamoe Technologies and contributors.
+ *
+ *  此源代码的使用受 GNU AFFERO GENERAL PUBLIC LICENSE version 3 许可证的约束, 可以在以下链接找到该许可证.
+ *  Use of this source code is governed by the GNU AGPLv3 license that can be found through the following link.
+ *
+ *  https://github.com/mamoe/mirai/blob/master/LICENSE
+ */
+
+package net.mamoe.mirai.internal.network.impl.netty
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import net.mamoe.mirai.event.events.BotEvent
+import net.mamoe.mirai.event.events.BotOfflineEvent
+import net.mamoe.mirai.event.events.BotReloginEvent
+import net.mamoe.mirai.event.nextEvent
+import net.mamoe.mirai.internal.network.handler.NetworkHandler
+import net.mamoe.mirai.internal.test.assertEventBroadcasts
+import net.mamoe.mirai.internal.test.runBlockingUnit
+import net.mamoe.mirai.utils.firstIsInstanceOrNull
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.TestInstance
+import java.io.IOException
+import kotlin.test.assertFailsWith
+import kotlin.test.assertNotNull
+
+@TestInstance(TestInstance.Lifecycle.PER_METHOD)
+internal class NettyBotNormalLoginTest : AbstractNettyNHTest() {
+    class CusLoginException(message: String?) : RuntimeException(message)
+
+    @Test
+    fun `test login fail`() = runBlockingUnit {
+        withSsoProcessor { throw CusLoginException("A") }
+        assertFailsWith<CusLoginException>("A") { bot.login() }
+    }
+
+    @Test
+    fun `test network broken`() = runBlockingUnit {
+        withSsoProcessor {
+            delay(1000)
+            channel.pipeline().fireExceptionCaught(IOException("TestNetworkBroken"))
+            delay(100000) // receive bits from "network"
+        }
+        assertFailsWith<IOException>("TestNetworkBroken") {
+            bot.login()
+        }
+    }
+
+    @Test
+    fun `test errors after logon`() = runBlockingUnit {
+        bot.login()
+        delay(1000)
+        assertEventBroadcasts<BotEvent>(-1) {
+            launch {
+                delay(1000)
+                channel.pipeline().fireExceptionCaught(CusLoginException("Net error"))
+            }
+            assertNotNull(
+                nextEvent<BotReloginEvent>(5000) { it.bot === bot }
+            )
+        }.let { events ->
+            assertFailsWith<CusLoginException>("Net error") {
+                throw events.firstIsInstanceOrNull<BotOfflineEvent.Dropped>()!!.cause!!
+            }
+        }
+        assertState(NetworkHandler.State.OK)
+    }
+}

mirai-core/src/commonTest/kotlin/test/events.kt

@@ -40,6 +40,8 @@ internal inline fun <reified T : Event> assertEventBroadcasts(times: Int = 1, bl
         listener.complete()
     }
 
+    if (times < 0) return receivedEvents.filterIsInstance<T>().cast()
+
     val actual = receivedEvents.filterIsInstance<T>().count()
     assertEquals(
         times,