Add network socket example for ktor-network and netty
This commit is contained in:
18 files changed
+271
-60
No files matched your search
@@ -0,0 +1,19 @@
|
||||
import org.jetbrains.kotlin.gradle.dsl.JvmTarget
|
||||
|
||||
kotlin {
|
||||
explicitApi()
|
||||
withSourcesJar()
|
||||
|
||||
linuxX64()
|
||||
linuxArm64()
|
||||
macosArm64()
|
||||
mingwX64()
|
||||
jvm { compilerOptions.jvmTarget = JvmTarget.JVM_1_8 }
|
||||
|
||||
sourceSets {
|
||||
jvmMain.dependencies {
|
||||
api(project(":protocol"))
|
||||
api(libs.ktor.network)
|
||||
}
|
||||
}
|
||||
}
|
||||
+10
@@ -0,0 +1,10 @@
|
||||
/*
|
||||
* Copyright © 2026 RTAkland
|
||||
* Author: RTAkland
|
||||
* Date: 2026/9/8
|
||||
*/
|
||||
|
||||
|
||||
package engines
|
||||
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
import org.jetbrains.kotlin.gradle.dsl.JvmTarget
|
||||
|
||||
kotlin {
|
||||
explicitApi()
|
||||
withSourcesJar()
|
||||
|
||||
jvm { compilerOptions.jvmTarget = JvmTarget.JVM_1_8 }
|
||||
|
||||
sourceSets {
|
||||
jvmMain.dependencies {
|
||||
api(project(":protocol"))
|
||||
api(libs.netty.handler)
|
||||
api(libs.netty.buffer)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,183 @@
|
||||
/*
|
||||
* Copyright © 2026 RTAkland
|
||||
* Author: RTAkland
|
||||
* Date: 2026/9/8
|
||||
*/
|
||||
|
||||
package engines
|
||||
|
||||
import cn.rtast.libmc.network.RawSocket
|
||||
import cn.rtast.libmc.network.ReadChannel
|
||||
import cn.rtast.libmc.network.SocketEngine
|
||||
import cn.rtast.libmc.network.WriteChannel
|
||||
import io.netty.bootstrap.Bootstrap
|
||||
import io.netty.buffer.ByteBuf
|
||||
import io.netty.buffer.Unpooled
|
||||
import io.netty.channel.*
|
||||
import io.netty.channel.nio.NioIoHandler
|
||||
import io.netty.channel.socket.SocketChannel
|
||||
import io.netty.channel.socket.nio.NioSocketChannel
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.suspendCancellableCoroutine
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import kotlinx.coroutines.withContext
|
||||
import kotlin.coroutines.resume
|
||||
import kotlin.coroutines.resumeWithException
|
||||
import io.netty.channel.Channel as NettyChannel
|
||||
import kotlinx.coroutines.channels.Channel as KotlinChannel
|
||||
|
||||
public class NettyNetworkSocketEngine : SocketEngine {
|
||||
override fun create(host: String, port: Int): RawSocket = NettyNetworkSocket(host, port)
|
||||
|
||||
private class NettyNetworkSocket(private val host: String, private val port: Int) : RawSocket {
|
||||
private val workerGroup: EventLoopGroup = MultiThreadIoEventLoopGroup(0, NioIoHandler.newFactory())
|
||||
private lateinit var channel: NettyChannel
|
||||
private val inboundQueue = KotlinChannel<ByteArray>(KotlinChannel.UNLIMITED)
|
||||
private lateinit var readChannel: NettyReadChannel
|
||||
private lateinit var writeChannel: NettyWriteChannel
|
||||
|
||||
override suspend fun connect() {
|
||||
val bootstrap = Bootstrap()
|
||||
.group(workerGroup)
|
||||
.channel(NioSocketChannel::class.java)
|
||||
.option(ChannelOption.TCP_NODELAY, true)
|
||||
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000)
|
||||
.handler(object : ChannelInitializer<SocketChannel>() {
|
||||
override fun initChannel(ch: SocketChannel) {
|
||||
ch.pipeline().addLast(object : ChannelInboundHandlerAdapter() {
|
||||
override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
|
||||
if (msg is ByteBuf) {
|
||||
try {
|
||||
val bytes = ByteArray(msg.readableBytes())
|
||||
msg.readBytes(bytes)
|
||||
inboundQueue.trySend(bytes)
|
||||
} finally {
|
||||
msg.release()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
|
||||
inboundQueue.close(cause)
|
||||
ctx.close()
|
||||
}
|
||||
|
||||
override fun channelInactive(ctx: ChannelHandlerContext) {
|
||||
inboundQueue.close()
|
||||
super.channelInactive(ctx)
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
|
||||
channel = withContext(Dispatchers.IO) {
|
||||
bootstrap.connect(host, port).suspendAwait()
|
||||
}
|
||||
|
||||
readChannel = NettyReadChannel(inboundQueue)
|
||||
writeChannel = NettyWriteChannel(channel)
|
||||
}
|
||||
|
||||
override fun openReadChannel(): ReadChannel = readChannel
|
||||
override fun openWriteChannel(): WriteChannel = writeChannel
|
||||
|
||||
override fun close() {
|
||||
if (::channel.isInitialized && channel.isOpen) {
|
||||
channel.close()
|
||||
}
|
||||
inboundQueue.close()
|
||||
workerGroup.shutdownGracefully()
|
||||
}
|
||||
}
|
||||
|
||||
private class NettyReadChannel(private val inboundQueue: KotlinChannel<ByteArray>) : ReadChannel {
|
||||
private var currentChunk: ByteArray? = null
|
||||
private var chunkOffset = 0
|
||||
|
||||
private suspend fun fetchNextChunk() {
|
||||
val next = inboundQueue.receiveCatching().getOrNull()
|
||||
?: throw IllegalStateException("Socket/Channel closed while reading")
|
||||
currentChunk = next
|
||||
chunkOffset = 0
|
||||
}
|
||||
|
||||
override suspend fun readByte(): Byte {
|
||||
while (currentChunk == null || chunkOffset >= currentChunk!!.size) {
|
||||
fetchNextChunk()
|
||||
}
|
||||
val chunk = currentChunk!!
|
||||
return chunk[chunkOffset++]
|
||||
}
|
||||
|
||||
override suspend fun readBytes(length: Int): ByteArray {
|
||||
val result = ByteArray(length)
|
||||
readFully(result, 0, length)
|
||||
return result
|
||||
}
|
||||
|
||||
override suspend fun readFully(out: ByteArray, start: Int, end: Int) {
|
||||
var written = start
|
||||
|
||||
while (written < end) {
|
||||
while (currentChunk == null || chunkOffset >= currentChunk!!.size) {
|
||||
fetchNextChunk()
|
||||
}
|
||||
val chunk = currentChunk!!
|
||||
val available = chunk.size - chunkOffset
|
||||
val toCopy = minOf(available, end - written)
|
||||
|
||||
chunk.copyInto(
|
||||
out,
|
||||
destinationOffset = written,
|
||||
startIndex = chunkOffset,
|
||||
endIndex = chunkOffset + toCopy
|
||||
)
|
||||
chunkOffset += toCopy
|
||||
written += toCopy
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class NettyWriteChannel(private val channel: NettyChannel) : WriteChannel {
|
||||
private val writeMutex = Mutex()
|
||||
|
||||
override suspend fun writeFully(value: ByteArray, startIndex: Int, endIndex: Int) {
|
||||
val length = endIndex - startIndex
|
||||
if (length <= 0) return
|
||||
|
||||
val nettyBuf = Unpooled.copiedBuffer(value, startIndex, length)
|
||||
|
||||
writeMutex.withLock {
|
||||
withContext(Dispatchers.IO) {
|
||||
channel.writeAndFlush(nettyBuf).suspendAwait()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun flush() {
|
||||
writeMutex.withLock {
|
||||
withContext(Dispatchers.IO) {
|
||||
channel.flush()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private suspend inline fun ChannelFuture.suspendAwait(): NettyChannel = suspendCancellableCoroutine { cont ->
|
||||
if (isDone) {
|
||||
if (isSuccess) cont.resume(channel())
|
||||
else cont.resumeWithException(cause() ?: RuntimeException("Netty operation failed"))
|
||||
return@suspendCancellableCoroutine
|
||||
}
|
||||
|
||||
addListener { future ->
|
||||
if (future.isSuccess) {
|
||||
cont.resume(channel())
|
||||
} else {
|
||||
cont.resumeWithException(future.cause() ?: RuntimeException("Netty operation failed"))
|
||||
}
|
||||
}
|
||||
cont.invokeOnCancellation { cancel(false) }
|
||||
}
|
||||
@@ -6,8 +6,7 @@ coroutines-test = "1.11.0"
|
||||
kotlinx-coroutines = "1.11.0"
|
||||
ktor-core = "3.5.2"
|
||||
cryptography-core = "0.6.0"
|
||||
kotlinx-atomicfu = "0.33.0"
|
||||
#kotlinx-serialization = "1.11.0"
|
||||
netty = "4.2.17.Final"
|
||||
|
||||
[libraries]
|
||||
kotlinx-io = { module = "org.jetbrains.kotlinx:kotlinx-io-core", version.ref = "kotlinx-io" }
|
||||
@@ -21,12 +20,9 @@ ktor-client-darwin = { module = "io.ktor:ktor-client-darwin", version.ref = "kto
|
||||
ktor-client-winhttp = { module = "io.ktor:ktor-client-winhttp", version.ref = "ktor-core" }
|
||||
ktor-client-curl = { module = "io.ktor:ktor-client-curl", version.ref = "ktor-core" }
|
||||
ktor-client-okhttp = { module = "io.ktor:ktor-client-okhttp", version.ref = "ktor-core" }
|
||||
kotlinx-atomicfu = { module = "org.jetbrains.kotlinx:atomicfu", version.ref = "kotlinx-atomicfu" }
|
||||
#kotlinx-serialization-core = { module = "org.jetbrains.kotlinx:kotlinx-serialization-core", version.ref = "kotlinx-serialization" }
|
||||
#kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serialization-json", version.ref = "kotlinx-serialization" }
|
||||
netty-handler = { module = "io.netty:netty-transport", version.ref = "netty" }
|
||||
netty-buffer = { module = "io.netty:netty-buffer", version.ref = "netty" }
|
||||
|
||||
[plugins]
|
||||
kotlin-multiplatform = { id = "org.jetbrains.kotlin.multiplatform", version.ref = "kotlin" }
|
||||
maven-publish = { id = "maven-publish" }
|
||||
kotlinx-atomicfu = { id = "org.jetbrains.kotlinx.atomicfu", version.ref = "kotlinx-atomicfu" }
|
||||
#kotlinx-serialization = { id = "org.jetbrains.kotlin.plugin.serialization", version.ref = "kotlin" }
|
||||
@@ -18,10 +18,6 @@ kotlin {
|
||||
|
||||
jvmMain.dependencies {}
|
||||
|
||||
// nativeMain.dependencies {
|
||||
// implementation(libs.ktor.network)
|
||||
// }
|
||||
|
||||
commonTest.dependencies {
|
||||
implementation(kotlin("test"))
|
||||
implementation(libs.kotlinx.coroutines.test)
|
||||
|
||||
@@ -7,8 +7,14 @@
|
||||
|
||||
package cn.rtast.libmc.network
|
||||
|
||||
import cn.rtast.libmc.crypto.ProtocolContextBuilder
|
||||
|
||||
public abstract class SocketContext {
|
||||
public abstract val engine: SocketEngine
|
||||
|
||||
public fun createSocket(host: String, port: Int): RawSocket = engine.create(host, port)
|
||||
}
|
||||
|
||||
public fun (ProtocolContextBuilder.() -> Unit).withCustom(
|
||||
block: ProtocolContextBuilder.() -> Unit,
|
||||
): ProtocolContextBuilder.() -> Unit = { this@withCustom.invoke(this); this.block() }
|
||||
@@ -16,29 +16,6 @@ kotlin {
|
||||
implementation(libs.cryptography.core)
|
||||
implementation(libs.cryptography.provider.optimal)
|
||||
implementation(libs.ktor.client.core)
|
||||
implementation(libs.ktor.network)
|
||||
}
|
||||
|
||||
jvmTest.dependencies {
|
||||
implementation(libs.ktor.client.okhttp)
|
||||
}
|
||||
|
||||
linuxTest.dependencies {
|
||||
implementation(libs.ktor.client.curl)
|
||||
}
|
||||
|
||||
mingwTest.dependencies {
|
||||
implementation(libs.ktor.client.winhttp)
|
||||
}
|
||||
|
||||
appleTest.dependencies {
|
||||
implementation(libs.ktor.client.darwin)
|
||||
}
|
||||
|
||||
commonTest.dependencies {
|
||||
implementation(kotlin("test"))
|
||||
implementation(project(":protocol"))
|
||||
implementation(libs.kotlinx.coroutines.test)
|
||||
}
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -7,7 +7,7 @@
|
||||
|
||||
@file:OptIn(DelicateCryptographyApi::class)
|
||||
|
||||
package cn.rtast.libmc.protocol.crypto
|
||||
package cn.rtast.libmc.protocol.context
|
||||
|
||||
import cn.rtast.libmc.crypto.NetworkCipher
|
||||
import dev.whyoleg.cryptography.DelicateCryptographyApi
|
||||
+2
-3
@@ -5,7 +5,7 @@
|
||||
*/
|
||||
|
||||
|
||||
package cn.rtast.libmc.protocol.crypto
|
||||
package cn.rtast.libmc.protocol.context
|
||||
|
||||
import cn.rtast.libmc.crypto.AuthenticationProvider
|
||||
import cn.rtast.libmc.crypto.ProtocolContextBuilder
|
||||
@@ -15,7 +15,7 @@ import io.ktor.client.*
|
||||
import io.ktor.client.request.*
|
||||
import io.ktor.http.*
|
||||
|
||||
private val httpClient = HttpClient()
|
||||
public val httpClient: HttpClient = HttpClient()
|
||||
|
||||
public val DefaultProtocolContext: ProtocolContextBuilder.() -> Unit = {
|
||||
rsaEncryptor = RSA1024Encryptor { key, data -> rsaEncrypt(key, data) }
|
||||
@@ -28,5 +28,4 @@ public val DefaultProtocolContext: ProtocolContextBuilder.() -> Unit = {
|
||||
}.status
|
||||
require(status == HttpStatusCode.NoContent)
|
||||
}
|
||||
socketEngine = KtorNetworkEngine()
|
||||
}
|
||||
+1
-1
@@ -7,7 +7,7 @@
|
||||
|
||||
@file:OptIn(DelicateCryptographyApi::class)
|
||||
|
||||
package cn.rtast.libmc.protocol.crypto
|
||||
package cn.rtast.libmc.protocol.context
|
||||
|
||||
import dev.whyoleg.cryptography.CryptographyProvider
|
||||
import dev.whyoleg.cryptography.DelicateCryptographyApi
|
||||
+1
-1
@@ -7,7 +7,7 @@
|
||||
|
||||
@file:OptIn(DelicateCryptographyApi::class)
|
||||
|
||||
package cn.rtast.libmc.protocol.crypto
|
||||
package cn.rtast.libmc.protocol.context
|
||||
|
||||
import dev.whyoleg.cryptography.DelicateCryptographyApi
|
||||
import dev.whyoleg.cryptography.algorithms.SHA1
|
||||
@@ -8,7 +8,7 @@ package test
|
||||
|
||||
import cn.rtast.libmc.packet.MinecraftPacket
|
||||
import cn.rtast.libmc.protocol.client.createMinecraftClient
|
||||
import cn.rtast.libmc.protocol.crypto.DefaultProtocolContext
|
||||
import cn.rtast.libmc.protocol.context.DefaultProtocolContext
|
||||
import kotlinx.coroutines.launch
|
||||
import org.junit.Test
|
||||
import java.io.File
|
||||
|
||||
@@ -24,6 +24,7 @@ kotlin {
|
||||
implementation(kotlin("test"))
|
||||
implementation(project(":protocol-context"))
|
||||
implementation(libs.kotlinx.coroutines.test)
|
||||
implementation(libs.ktor.network)
|
||||
}
|
||||
|
||||
jvmTest.dependencies {
|
||||
|
||||
+2
-2
@@ -97,9 +97,9 @@ public fun createMinecraftClient(
|
||||
accessToken: String?,
|
||||
parentJob: Job? = null,
|
||||
ioDispatcher: CoroutineDispatcher = Dispatchers.IO,
|
||||
contextBuilder: ProtocolContextBuilder.() -> Unit,
|
||||
context: ProtocolContextBuilder.() -> Unit,
|
||||
): MinecraftClient {
|
||||
val context = ProtocolContextBuilder(accessToken != null).apply(contextBuilder).build()
|
||||
val context = ProtocolContextBuilder(accessToken != null).apply(context).build()
|
||||
return MinecraftClient(
|
||||
host = host,
|
||||
port = port,
|
||||
|
||||
+5
-5
@@ -5,7 +5,7 @@
|
||||
*/
|
||||
|
||||
|
||||
package cn.rtast.libmc.protocol.crypto
|
||||
package test
|
||||
|
||||
import cn.rtast.libmc.network.RawSocket
|
||||
import cn.rtast.libmc.network.ReadChannel
|
||||
@@ -17,11 +17,11 @@ import io.ktor.utils.io.*
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.IO
|
||||
|
||||
public class KtorNetworkEngine : SocketEngine {
|
||||
class KtorNetworkEngine : SocketEngine {
|
||||
override fun create(host: String, port: Int): RawSocket = KtorNetworkSocket(host, port)
|
||||
}
|
||||
|
||||
public class KtorNetworkSocket(private val host: String, private val port: Int) : RawSocket {
|
||||
class KtorNetworkSocket(private val host: String, private val port: Int) : RawSocket {
|
||||
private val sm = SelectorManager(Dispatchers.IO)
|
||||
private lateinit var socket: Socket
|
||||
|
||||
@@ -34,14 +34,14 @@ public class KtorNetworkSocket(private val host: String, private val port: Int)
|
||||
}
|
||||
}
|
||||
|
||||
public class KtorReadChannel(private val readChannel: ByteReadChannel) : ReadChannel {
|
||||
class KtorReadChannel(private val readChannel: ByteReadChannel) : ReadChannel {
|
||||
override suspend fun readByte(): Byte = readChannel.readByte()
|
||||
override suspend fun readBytes(length: Int): ByteArray = readChannel.readByteArray(length)
|
||||
override suspend fun readFully(out: ByteArray, start: Int, end: Int): Unit =
|
||||
readChannel.readFully(out, start, end)
|
||||
}
|
||||
|
||||
public class KtorWriteChannel(private val writeChannel: ByteWriteChannel) : WriteChannel {
|
||||
class KtorWriteChannel(private val writeChannel: ByteWriteChannel) : WriteChannel {
|
||||
override suspend fun writeFully(value: ByteArray, startIndex: Int, endIndex: Int) {
|
||||
writeChannel.writeFully(value, startIndex, endIndex)
|
||||
}
|
||||
@@ -7,9 +7,10 @@
|
||||
|
||||
package test
|
||||
|
||||
import cn.rtast.libmc.network.withCustom
|
||||
import cn.rtast.libmc.packet.ClientboundUnknownPacket
|
||||
import cn.rtast.libmc.protocol.client.createMinecraftClient
|
||||
import cn.rtast.libmc.protocol.crypto.DefaultProtocolContext
|
||||
import cn.rtast.libmc.protocol.context.DefaultProtocolContext
|
||||
import cn.rtast.libmc.protocol.packet.play.clientbound.ClientboundPlayerChatMessagePacket
|
||||
import cn.rtast.libmc.protocol.packet.play.serverbound.ServerboundChatMessagePacket
|
||||
import cn.rtast.libmc.protocol.util.generateOfflineUuid
|
||||
@@ -63,7 +64,7 @@ class TestClientTestInJvm {
|
||||
Uuid.parse("bb033844-e68e-4909-a636-1a5d1821ddc4"),
|
||||
// null,
|
||||
accessToken,
|
||||
contextBuilder = DefaultProtocolContext
|
||||
context = DefaultProtocolContext
|
||||
)
|
||||
// cli.onPacket<ClientboundSystemChatMessagePacket> { println(it) }
|
||||
// cli.onPacket<ClientboundLoginSuccessPacket> { println(it) }
|
||||
@@ -81,8 +82,11 @@ class TestClientTestInJvm {
|
||||
"11",
|
||||
generateOfflineUuid("11"),
|
||||
null,
|
||||
contextBuilder = DefaultProtocolContext
|
||||
context = DefaultProtocolContext.withCustom {
|
||||
socketEngine = KtorNetworkEngine()
|
||||
}
|
||||
)
|
||||
|
||||
// cli.on { packet, direction ->
|
||||
// if (packet !is ClientboundWaypointPacket)
|
||||
// println("$direction -> $packet")
|
||||
|
||||
+11
-7
@@ -4,12 +4,16 @@ plugins {
|
||||
|
||||
rootProject.name = "libmc"
|
||||
|
||||
includeSubModule(":common")
|
||||
includeSubModule(":protocol")
|
||||
includeSubModule(":protocol-context")
|
||||
includeSubModule(":nbt")
|
||||
//includeSubModule(":snbt")
|
||||
includeSubModule("common")
|
||||
includeSubModule("protocol")
|
||||
includeSubModule("protocol-context")
|
||||
includeSubModule("nbt")
|
||||
includeSubModule("snbt")
|
||||
|
||||
fun includeSubModule(name: String, path: String? = null) = include(name).also {
|
||||
project(name).projectDir = file(path ?: "libmc-${name.removePrefix(":")}")
|
||||
//includeSubModule("protocol-engine-netty", path = "libmc-network-engines/netty")
|
||||
//includeSubModule("protocol-engine-ktor-network", path = "libmc-network-engines/ktor-network")
|
||||
|
||||
fun includeSubModule(name: String, path: String? = null) {
|
||||
include(":$name")
|
||||
project(":$name").projectDir = file(path ?: "libmc-$name")
|
||||
}
|
||||
Reference in New Issue
Block a user