Decode packets in netty event loop group (#1500)

* Decode packets in netty event loop group * Update mirai-core/src/commonTest/kotlin/network/framework/AbstractNettyNHTest.kt Co-authored-by: Him188 <Him188@mamoe.net> Co-authored-by: Him188 <Him188@mamoe.net>
mamoe · Sep 1, 2021 · 88b66d7 · 88b66d7
1 parent 4d76aa8
commit 88b66d7
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 5 deletions.
diff --git a/mirai-core/src/commonMain/kotlin/network/impl/netty/NettyNetworkHandler.kt b/mirai-core/src/commonMain/kotlin/network/impl/netty/NettyNetworkHandler.kt
@@ -113,12 +113,18 @@ internal open class NettyNetworkHandler(
             .addLast(RawIncomingPacketCollector(decodePipeline))
     }
 
+    protected open fun createDummyDecodePipeline() = PacketDecodePipeline(this@NettyNetworkHandler.coroutineContext)
+
     // can be overridden for tests
-    protected open suspend fun createConnection(decodePipeline: PacketDecodePipeline): NettyChannel {
+    protected open suspend fun createConnection(): NettyChannel {
         packetLogger.debug { "Connecting to $address" }
 
         val contextResult = CompletableDeferred<NettyChannel>()
         val eventLoopGroup = NioEventLoopGroup()
+        val decodePipeline = PacketDecodePipeline(
+            this@NettyNetworkHandler.coroutineContext
+                .plus(eventLoopGroup.asCoroutineDispatcher())
+        )
 
         val future = Bootstrap().group(eventLoopGroup)
             .channel(NioSocketChannel::class.java)
@@ -159,8 +165,6 @@ internal open class NettyNetworkHandler(
         return contextResult.await()
     }
 
-    protected val decodePipeline = PacketDecodePipeline(this@NettyNetworkHandler.coroutineContext)
-
     protected inner class PacketDecodePipeline(parentContext: CoroutineContext) :
         CoroutineScope by parentContext.childScope() {
         private val packetCodec: PacketCodec by lazy { context[PacketCodec] }
@@ -241,7 +245,7 @@ internal open class NettyNetworkHandler(
         private val collectiveExceptions: ExceptionCollector,
     ) : NettyState(State.CONNECTING) {
         private val connection = async {
-            createConnection(decodePipeline)
+            createConnection()
         }
 
         @Suppress("JoinDeclarationAndAssignment")

diff --git a/mirai-core/src/commonTest/kotlin/network/framework/AbstractNettyNHTest.kt b/mirai-core/src/commonTest/kotlin/network/framework/AbstractNettyNHTest.kt
@@ -33,7 +33,10 @@ internal abstract class TestNettyNH(
     address: SocketAddress,
 ) : NettyNetworkHandler(context, address), ITestNetworkHandler {
 
-    abstract override suspend fun createConnection(decodePipeline: PacketDecodePipeline): Channel
+    protected abstract suspend fun createConnection(decodePipeline: PacketDecodePipeline): Channel
+    final override suspend fun createConnection(): Channel {
+        return createConnection(createDummyDecodePipeline())
+    }
 
     override fun setStateClosed(exception: Throwable?): NetworkHandlerSupport.BaseStateImpl? {
         return setState { StateClosed(exception) }