feat: introduce persistent player data tracking and binary tag handling

qodana.yaml

@@ -29,3 +29,15 @@ projectJDK: "21" #(Applied in CI/CD pipeline)
 
 #Specify Qodana linter for analysis (Applied in CI/CD pipeline)
 linter: jetbrains/qodana-jvm:2024.3
+
+exclude:
+  - name: All
+    paths:
+      - surf-cloud-test-plugin
+      - api
+      - cert_generation
+      - cert_new
+      - Writerside
+
+include:
+  - name: IncorrectFormatting

settings.gradle.kts

@@ -27,16 +27,21 @@ include("surf-cloud-core:surf-cloud-core-client")
 findProject(":surf-cloud-core:surf-cloud-core-client")?.name = "surf-cloud-core-client"
 
 include("surf-cloud-api:surf-cloud-api-client:surf-cloud-api-client-common")
-findProject(":surf-cloud-api:surf-cloud-api-client:surf-cloud-api-client-common")?.name = "surf-cloud-api-client-common"
+findProject(":surf-cloud-api:surf-cloud-api-client:surf-cloud-api-client-common")?.name =
+    "surf-cloud-api-client-common"
 
 include("surf-cloud-api:surf-cloud-api-client:surf-cloud-api-client-paper")
-findProject(":surf-cloud-api:surf-cloud-api-client:surf-cloud-api-client-paper")?.name = "surf-cloud-api-client-paper"
+findProject(":surf-cloud-api:surf-cloud-api-client:surf-cloud-api-client-paper")?.name =
+    "surf-cloud-api-client-paper"
 
 include("surf-cloud-api:surf-cloud-api-client:surf-cloud-api-client-velocity")
-findProject(":surf-cloud-api:surf-cloud-api-client:surf-cloud-api-client-velocity")?.name = "surf-cloud-api-client-velocity"
+findProject(":surf-cloud-api:surf-cloud-api-client:surf-cloud-api-client-velocity")?.name =
+    "surf-cloud-api-client-velocity"
 
 include("surf-cloud-bom")
 
 if (!ci) {
     include(":surf-cloud-test-plugin:surf-cloud-test-standalone")
-}
+    include(":surf-cloud-test-plugin:surf-cloud-test-core")
+    include("surf-cloud-test-plugin:surf-cloud-test-paper")
+}

surf-cloud-api/surf-cloud-api-common/src/main/kotlin/dev/slne/surf/cloud/api/common/internal/BinaryTagTypeProxy.kt

@@ -0,0 +1,19 @@
+package dev.slne.surf.cloud.api.common.internal
+
+import dev.slne.surf.cloud.api.common.util.annotation.InternalApi
+import dev.slne.surf.surfapi.core.api.reflection.*
+import net.kyori.adventure.nbt.BinaryTag
+import net.kyori.adventure.nbt.BinaryTagType
+
+@InternalApi
+@SurfProxy(BinaryTagType::class)
+internal interface BinaryTagTypeProxy {
+
+    @Static
+    @Field("TYPES", Field.Type.GETTER)
+    fun getTypes(): List<BinaryTagType<out BinaryTag>>
+
+    companion object {
+        internal val instance = surfReflection.createProxy<BinaryTagTypeProxy>()
+    }
+}

surf-cloud-api/surf-cloud-api-common/src/main/kotlin/dev/slne/surf/cloud/api/common/netty/network/codec/ByteBufCodecs.kt

@@ -3,14 +3,27 @@ package dev.slne.surf.cloud.api.common.netty.network.codec
 import dev.slne.surf.cloud.api.common.netty.protocol.buffer.*
 import dev.slne.surf.cloud.api.common.netty.protocol.buffer.types.Utf8String
 import dev.slne.surf.cloud.api.common.util.ByIdMap
+import dev.slne.surf.cloud.api.common.util.ByIdMap.OutOfBoundsStrategy
 import dev.slne.surf.cloud.api.common.util.createUnresolvedInetSocketAddress
 import io.netty.buffer.ByteBuf
+import io.netty.buffer.ByteBufInputStream
+import io.netty.buffer.ByteBufOutputStream
+import io.netty.handler.codec.DecoderException
+import io.netty.handler.codec.EncoderException
+import it.unimi.dsi.fastutil.io.FastBufferedInputStream
+import it.unimi.dsi.fastutil.io.FastBufferedOutputStream
+import it.unimi.dsi.fastutil.objects.ObjectArrayList
 import net.kyori.adventure.key.Key
+import net.kyori.adventure.nbt.BinaryTag
 import net.kyori.adventure.nbt.BinaryTagIO
+import net.kyori.adventure.nbt.BinaryTagType
+import net.kyori.adventure.nbt.BinaryTagTypes
 import net.kyori.adventure.sound.Sound
 import net.kyori.adventure.text.serializer.gson.GsonComponentSerializer
 import java.io.ByteArrayInputStream
 import java.io.ByteArrayOutputStream
+import java.io.DataInputStream
+import java.io.DataOutputStream
 import java.math.BigDecimal
 import java.math.BigInteger
 import java.math.MathContext
@@ -21,10 +34,15 @@ import java.time.Instant
 import java.time.ZoneId
 import java.time.ZonedDateTime
 import java.util.*
+import java.util.zip.GZIPInputStream
+import java.util.zip.GZIPOutputStream
+import kotlin.math.min
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.milliseconds
 
 object ByteBufCodecs {
+    const val MAX_INITIAL_COLLECTION_SIZE = 65536
+
     val BOOLEAN_CODEC = streamCodec(ByteBuf::writeBoolean, ByteBuf::readBoolean)
     val BYTE_CODEC = streamCodec({ buf, byte -> buf.writeByte(byte.toInt()) }, ByteBuf::readByte)
     val SHORT_CODEC =
@@ -59,7 +77,7 @@ object ByteBufCodecs {
     private val SOUND_SOURCE_BY_ID = ByIdMap.continuous(
         { it.ordinal },
         Sound.Source.entries.toTypedArray(),
-        ByIdMap.OutOfBoundsStrategy.ZERO
+        OutOfBoundsStrategy.ZERO
     )
     val SOUND_CODEC = streamCodecComposite(
         KEY_CODEC,
@@ -138,6 +156,85 @@ object ByteBufCodecs {
         BigDecimal(unscaledValue, scale, MathContext(precision))
     }
 
+    private val TAG_TYPES = arrayOf(
+        BinaryTagTypes.END,
+        BinaryTagTypes.BYTE,
+        BinaryTagTypes.SHORT,
+        BinaryTagTypes.INT,
+        BinaryTagTypes.LONG,
+        BinaryTagTypes.FLOAT,
+        BinaryTagTypes.DOUBLE,
+        BinaryTagTypes.BYTE_ARRAY,
+        BinaryTagTypes.STRING,
+        BinaryTagTypes.LIST,
+        BinaryTagTypes.COMPOUND,
+        BinaryTagTypes.INT_ARRAY,
+        BinaryTagTypes.LONG_ARRAY,
+    )
+
+    private val BINARY_TAG_BY_ID = ByIdMap.continuous(
+        { it.id().toInt() },
+        TAG_TYPES,
+        OutOfBoundsStrategy.DECODE_ERROR
+    )
+
+    val BINARY_TAG_CODEC: StreamCodec<ByteBuf, BinaryTag> = streamCodec({ buf, tag ->
+        val type = tag.type() as BinaryTagType<BinaryTag>
+        buf.writeByte(type.id().toInt())
+
+        val tmp = buf.alloc().buffer()
+        try {
+            DataOutputStream(FastBufferedOutputStream(ByteBufOutputStream(tmp))).use { out ->
+                type.write(tag, out)
+            }
+            val length = tmp.readableBytes()
+            buf.writeVarInt(length)
+            buf.writeBytes(tmp, tmp.readerIndex(), length)
+        } finally {
+            tmp.release()
+        }
+
+    }, { buf ->
+        val type = BINARY_TAG_BY_ID(buf.readByte().toInt())
+        val len = buf.readVarInt()
+        val slice = buf.readRetainedSlice(len)
+
+        try {
+            DataInputStream(FastBufferedInputStream(ByteBufInputStream(slice))).use { input ->
+                type.read(input)
+            }
+        } finally {
+            slice.release()
+        }
+    })
+
+    val BINARY_TAG_CODEC_COMPRESSED: StreamCodec<ByteBuf, BinaryTag> = streamCodec({ buf, tag ->
+        val type = tag.type() as BinaryTagType<BinaryTag>
+        buf.writeByte(type.id().toInt())
+        val temp = buf.alloc().buffer()
+
+        try {
+            DataOutputStream(FastBufferedOutputStream(GZIPOutputStream(ByteBufOutputStream(temp)))).use {
+                type.write(tag, it)
+            }
+            buf.writeVarInt(temp.readableBytes())
+            buf.writeBytes(temp, temp.readerIndex(), temp.readableBytes())
+        } finally {
+            temp.release()
+        }
+    }, { buf ->
+        val type = BINARY_TAG_BY_ID(buf.readByte().toInt())
+        val length = buf.readVarInt()
+        val slice = buf.readRetainedSlice(length)
+
+        try {
+            DataInputStream(FastBufferedInputStream(GZIPInputStream(ByteBufInputStream(slice)))).use {
+                type.read(it)
+            }
+        } finally {
+            slice.release()
+        }
+    })
 
     fun <E : Enum<E>> enumStreamCodec(clazz: Class<E>): StreamCodec<ByteBuf, E> =
         streamCodec(ByteBuf::writeEnum) { it.readEnum(clazz) }
@@ -155,4 +252,67 @@ object ByteBufCodecs {
             buf.writeVarInt(idGetter(value))
         }
     }
+
+    fun readCount(buffer: ByteBuf, maxSize: Int): Int {
+        val count = buffer.readVarInt()
+        if (count > maxSize) {
+            throw DecoderException("$count elements exceeded max size of: $maxSize")
+        } else {
+            return count
+        }
+    }
+
+    fun writeCount(buffer: ByteBuf, count: Int, maxSize: Int) {
+        if (count > maxSize) {
+            throw EncoderException("$count elements exceeded max size of: $maxSize")
+        } else {
+            buffer.writeVarInt(count)
+        }
+    }
+
+    fun <B : ByteBuf, V, C : MutableCollection<V>> collection(
+        factory: (Int) -> C,
+        codec: StreamCodec<in B, V>,
+        maxSize: Int = Int.MAX_VALUE
+    ): StreamCodec<B, C> = object : StreamCodec<B, C> {
+        override fun decode(buf: B): C {
+            val count = readCount(buf, maxSize)
+            val collection = factory(min(count, MAX_INITIAL_COLLECTION_SIZE))
+
+            repeat(count) {
+                collection.add(codec.decode(buf))
+            }
+
+            return collection
+        }
+
+        override fun encode(buf: B, value: C) {
+            writeCount(buf, value.size, maxSize)
+
+            for (element in value) {
+                codec.encode(buf, element)
+            }
+        }
+    }
+
+    fun <B : ByteBuf, V, C : MutableCollection<V>> collection(factory: (Int) -> C): CodecOperation<B, V, C> {
+        return CodecOperation { size -> collection(factory, size) }
+    }
+
+
+    fun <B : ByteBuf, V> list(): CodecOperation<B, V, MutableList<V>> {
+        return CodecOperation { size -> collection(::ObjectArrayList, size) }
+    }
+
+    fun <B : ByteBuf, V> list(maxSize: Int): CodecOperation<B, V, MutableList<V>> {
+        return CodecOperation { size -> collection(::ObjectArrayList, size, maxSize) }
+    }
+
+    @Suppress("NOTHING_TO_INLINE")
+    private inline fun decodeError(message: Any): Nothing =
+        throw DecoderException(message.toString())
+
+    @Suppress("NOTHING_TO_INLINE")
+    private inline fun encodeError(message: Any): Nothing =
+        throw EncoderException(message.toString())
 }

surf-cloud-api/surf-cloud-api-common/src/main/kotlin/dev/slne/surf/cloud/api/common/netty/packet/RespondingNettyPacket.kt

@@ -10,7 +10,9 @@ import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.withTimeout
 import kotlinx.coroutines.withTimeoutOrNull
+import java.lang.ref.WeakReference
 import java.util.*
+import kotlin.properties.Delegates
 import kotlin.time.Duration
 import kotlin.time.Duration.Companion.seconds
 
@@ -36,14 +38,18 @@ abstract class RespondingNettyPacket<P : ResponseNettyPacket> : NettyPacket() {
      * A deferred response object for awaiting the associated response.
      */
     @InternalApi
-    val response = CompletableDeferred<P>()
+    val response by lazy { CompletableDeferred<P>() }
 
     /**
      * The connection through which the response will be sent.
      */
     @InternalApi
-    lateinit var responseConnection: Connection
+    private var responseConnection by Delegates.notNull<WeakReference<Connection>>()
 
+    @InternalApi
+    fun initResponseConnection(connection: Connection) {
+        responseConnection = WeakReference(connection)
+    }
 
     /**
      * Fires the packet and awaits its response within the specified timeout.
@@ -109,6 +115,19 @@ abstract class RespondingNettyPacket<P : ResponseNettyPacket> : NettyPacket() {
     fun respond(packet: P) {
         packet.responseTo = uniqueSessionId
             ?: error("Responding packet has no session id. Are you sure it was sent?")
+
+        val responseConnection = responseConnection.get()
+        if (responseConnection == null) {
+            log.atWarning()
+                .log("Cannot respond to packet ${this::class.simpleName} with session ID $uniqueSessionId: original connection has been garbage collected")
+
+            if ((::response.getDelegate() as Lazy<*>).isInitialized()) {
+                response.completeExceptionally(IllegalStateException("Original connection has been garbage collected"))
+            }
+
+            return
+        }
+
         responseConnection.send(packet)
     }
 

surf-cloud-api/surf-cloud-api-common/src/main/kotlin/dev/slne/surf/cloud/api/common/player/CloudPlayer.kt

@@ -4,6 +4,7 @@
 import dev.slne.surf.cloud.api.common.netty.network.codec.kotlinx.cloud.CloudPlayerSerializer
 import dev.slne.surf.cloud.api.common.netty.network.codec.streamCodec
 import dev.slne.surf.cloud.api.common.player.ppdc.PersistentPlayerDataContainer
+import dev.slne.surf.cloud.api.common.player.ppdc.PersistentPlayerDataContainerView
 import dev.slne.surf.cloud.api.common.player.teleport.TeleportCause
 import dev.slne.surf.cloud.api.common.player.teleport.TeleportFlag
 import dev.slne.surf.cloud.api.common.player.teleport.WorldLocation
@@ -29,7 +30,7 @@
  * it enables sending messages or components to the player.
  */
 @Serializable(with = CloudPlayerSerializer::class)
-interface CloudPlayer : Audience, OfflineCloudPlayer { // TODO: conversation but done correctly?
+interface CloudPlayer : Audience, OfflineCloudPlayer {
     val name: String
 
     override suspend fun latestIpAddress(): Inet4Address
@@ -55,13 +56,19 @@
     fun isAfk(): Boolean
     suspend fun currentSessionDuration(): Duration
 
+    val persistentData: PersistentPlayerDataContainerView
+
     /**
      * Performs modifications on the player's persistent data container.
      *
-     * @param block A suspending block to modify the persistent data container.
+     * @param block A block to modify the persistent data container.
      * @return The result of the block execution.
      */
-    suspend fun <R> withPersistentData(block: PersistentPlayerDataContainer.() -> R): R
+    fun <R> editPdc(block: PersistentPlayerDataContainer.() -> R): R
+
+    @Deprecated("Use non-suspending editPdc method instead", ReplaceWith("editPdc(block)"))
+    suspend fun <R> withPersistentData(block: PersistentPlayerDataContainer.() -> R): R =
+        editPdc(block)
 
     /**
      * Connects the player to a specified server.
@@ -76,7 +83,7 @@
      *
      * @param group The server group name.
      * @param server The target server name.
      * @return A [ConnectionResult] indicating the result of the connection attempt.
      */
     suspend fun connectToServer(group: String, server: String): ConnectionResultEnum
 
@@ -84,7 +91,7 @@
      * Connects the player to the server with the lowest player count in the specified group.
      *
      * @param group The server group name.
      * @return A [ConnectionResult] indicating the result of the connection attempt.
      */
     suspend fun connectToServer(group: String): ConnectionResultEnum
 
@@ -92,7 +99,7 @@
      * Connects the player to the specified server or places them in a queue if unavailable.
      *
      * @param server The target server to connect to.
      * @return A [ConnectionResult] indicating the result of the connection attempt.
      */
     suspend fun connectToServerOrQueue(
         server: CloudServer,
@@ -104,7 +111,7 @@
      *
      * @param group The server group name.
      * @param server The target server name.
      * @return A [ConnectionResult] indicating the result of the connection attempt.
      */
     suspend fun connectToServerOrQueue(
         group: String,
@@ -116,7 +123,7 @@
      * Connects the player to the server with the lowest player count in a group or queues them if unavailable.
      *
      * @param group The server group name.
      * @return A [ConnectionResult] indicating the result of the connection attempt.
      */
     suspend fun connectToServerOrQueue(
         group: String,
@@ -212,7 +219,7 @@
     val isSuccess: Boolean = false
 ) {
     @Serializable
     object SUCCESS : ConnectionResultEnum(
         buildText { success("Du hast dich erfolgreich Verbunden.") },
         isSuccess = true
     )
@@ -227,57 +234,57 @@
     )
 
     @Serializable
     object SERVER_FULL : ConnectionResultEnum(
         buildText { error("Der Server ist voll.") }
     )
 
     @Serializable
     object CATEGORY_FULL : ConnectionResultEnum(
         buildText { error("Die Kategorie ist voll.") }
     )
 
     @Serializable
     object SERVER_OFFLINE : ConnectionResultEnum(
         buildText { error("Der Server ist offline.") }
     )
 
     @Serializable
     object ALREADY_CONNECTED : ConnectionResultEnum(
         buildText { error("Du bist bereits mit diesem Server verbunden.") }
     )
 
     @Serializable
     object CANNOT_SWITCH_PROXY : ConnectionResultEnum(
         buildText { error("Du kannst nicht zu diesem Server wechseln, da dieser unter einem anderen Proxy läuft.") }
     )
 
     @Serializable
     object OTHER_SERVER_CANNOT_ACCEPT_TRANSFER_PACKET : ConnectionResultEnum(
         buildText { error("Der Server kann das Transfer-Paket nicht akzeptieren.") }
     )
 
     @Serializable
     object CANNOT_COMMUNICATE_WITH_PROXY : ConnectionResultEnum(
         buildText { error("Der Proxy kann nicht erreicht werden.") }
     )
 
     @Serializable
     object CONNECTION_IN_PROGRESS : ConnectionResultEnum(
         buildText { error("Du versuchst bereits eine Verbindung zu einem Server herzustellen.") }
     )
 
     @Serializable
     object CONNECTION_CANCELLED : ConnectionResultEnum(
         buildText { error("Die Verbindung wurde abgebrochen.") }
     )
 
     @Serializable
     object SERVER_DISCONNECTED : ConnectionResultEnum(
         buildText { error("Der Server hat die Verbindung getrennt.") }
     )
 
     @Serializable
     object CANNOT_CONNECT_TO_PROXY : ConnectionResultEnum(
         buildText { error("Der Proxy kann nicht erreicht werden.") }
     )
 
@@ -316,7 +323,7 @@
     })
 
     @Serializable
     object DISCONNECTED : ConnectionResultEnum(
         buildText {
             error("Du hast die Verbindung getrennt.")
         }
@@ -334,7 +341,7 @@
     )
 
     @Serializable
     object SERVER_SWITCHED : ConnectionResultEnum(
         buildText {
             error("Du bist auf einen anderen Server gewechselt.")
         }

surf-cloud-api/surf-cloud-api-common/src/main/kotlin/dev/slne/surf/cloud/api/common/player/ppdc/ListPersistentPlayerDataType.kt

@@ -6,7 +6,7 @@
  * @param P The primitive type of the list elements.
  * @param C The complex type of the list elements.
  */
-interface ListPersistentPlayerDataType<P : Any, C>: PersistentPlayerDataType<List<P>, C> {
+interface ListPersistentPlayerDataType<P : Any, C>: PersistentPlayerDataType<List<P>, List<C>> {
 
     /**
      * The data type of the elements in the list.