github-repos/mirai
1
0
Fork 0
mirror of https://github.com/mamoe/mirai.git synced 2025-01-21 10:42:25 +08:00
Code Issues Packages Projects Releases Wiki Activity

Extract netty utils

Browse Source
This commit is contained in:
Him188 2021-04-16 21:12:07 +08:00
parent 90f5e27b5f
commit 1d053bd1bd
3 changed files with 43 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
mirai-core/src/commonMain/kotlin/network/handler/SelectorNetworkHandler.kt
View File

@ -91,7 +91,6 @@ internal abstract class AbstractKeepAliveNetworkHandlerSelector<H : NetworkHandl
final override fun getResumedInstance(): H? = current.value
// TODO: 2021/4/16 add test for AbstractKeepAliveNetworkHandlerSelector
final override tailrec suspend fun awaitResumeInstance(): H {
val current = getResumedInstance()
return if (current != null) {

31
mirai-core/src/commonMain/kotlin/network/handler/impl/netty/NettyNetworkHandler.kt
View File

@ -11,8 +11,10 @@ package net.mamoe.mirai.internal.network.handler.impl.netty
import io.netty.bootstrap.Bootstrap
import io.netty.buffer.ByteBuf
import io.netty.buffer.ByteBufInputStream
import io.netty.channel.*
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter
import io.netty.channel.ChannelInitializer
import io.netty.channel.SimpleChannelInboundHandler
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.NioSocketChannel
@ -22,7 +24,6 @@ import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.sendBlocking
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.consumeAsFlow
import kotlinx.io.core.ByteReadPacket
import net.mamoe.mirai.internal.network.handler.NetworkHandler
import net.mamoe.mirai.internal.network.handler.NetworkHandlerContext
import net.mamoe.mirai.internal.network.handler.impl.NetworkHandlerSupport
@ -208,27 +209,3 @@ internal class NettyNetworkHandler(
override fun initialState(): BaseStateImpl = StateInitialized()
}
internal suspend fun ChannelFuture.awaitKt(): ChannelFuture {
suspendCancellableCoroutine<Unit> { cont ->
cont.invokeOnCancellation {
channel().close()
}
addListener { f ->
if (f.isSuccess) {
cont.resumeWith(Result.success(Unit))
} else {
cont.resumeWith(Result.failure(f.cause()))
}
}
}
return this
}
// TODO: 2021/4/14 Add test for toReadPacket
private fun ByteBuf.toReadPacket(): ByteReadPacket {
val buf = this
return buildPacket {
ByteBufInputStream(buf).withUse { copyTo(outputStream()) }
}
}

39
mirai-core/src/commonMain/kotlin/network/handler/impl/netty/nettyUtils.kt Normal file
View File

@ -0,0 +1,39 @@
/*
* 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.handler.impl.netty
import io.netty.buffer.ByteBuf
import io.netty.buffer.ByteBufInputStream
import io.netty.channel.ChannelFuture
import kotlinx.io.core.ByteReadPacket
internal suspend fun ChannelFuture.awaitKt(): ChannelFuture {
suspendCancellableCoroutine<Unit> { cont ->
cont.invokeOnCancellation {
channel().close()
}
addListener { f ->
if (f.isSuccess) {
cont.resumeWith(Result.success(Unit))
} else {
cont.resumeWith(Result.failure(f.cause()))
}
}
}
return this
}
internal fun ByteBuf.toReadPacket(): ByteReadPacket {
val buf = this
return buildPacket {
ByteBufInputStream(buf).withUse { copyTo(outputStream()) }
}
}