Update matter registry networking

src/main/kotlin/ru/dbotthepony/mc/otm/core/util/FriendlyStreams.kt

@@ -53,14 +53,6 @@ fun InputStream.readBigDecimal(): BigDecimal {
 	return BigDecimal(BigInteger(bytes), scale)
 }
 
-fun InputStream.readBinaryComponent(): Component? {
-	return Component.Serializer.fromJson(readBinaryJson())
-}
-
-fun OutputStream.writeBinaryComponent(component: Component) {
-	writeBinaryJson(Component.Serializer.toJsonTree(component))
-}
-
 fun <S : OutputStream, V> S.writeCollection(collection: Collection<V>, writer: S.(V) -> Unit) {
 	writeVarIntLE(collection.size)
 
@@ -246,12 +238,3 @@ fun OutputStream.writeBinaryString(input: String) {
 	writeVarIntLE(bytes.size)
 	write(bytes)
 }
-
-fun InputStream.readResourceLocation(): ResourceLocation {
-	return ResourceLocation(readBinaryString(), readBinaryString())
-}
-
-fun OutputStream.writeResourceLocation(value: ResourceLocation) {
-	writeBinaryString(value.namespace)
-	writeBinaryString(value.path)
-}

src/main/kotlin/ru/dbotthepony/mc/otm/matter/MatterManager.kt

@@ -11,6 +11,7 @@ import com.google.gson.JsonSyntaxException
 import com.mojang.brigadier.arguments.StringArgumentType
 import com.mojang.brigadier.context.CommandContext
 import com.mojang.serialization.JsonOps
+import io.netty.buffer.ByteBufAllocator
 import it.unimi.dsi.fastutil.io.FastByteArrayInputStream
 import it.unimi.dsi.fastutil.io.FastByteArrayOutputStream
 import it.unimi.dsi.fastutil.objects.Object2BooleanFunction
@@ -26,8 +27,10 @@ import it.unimi.dsi.fastutil.objects.ReferenceOpenHashSet
 import net.minecraft.ChatFormatting
 import net.minecraft.commands.CommandSourceStack
 import net.minecraft.commands.Commands
+import net.minecraft.core.RegistryAccess
 import net.minecraft.core.registries.BuiltInRegistries
 import net.minecraft.network.FriendlyByteBuf
+import net.minecraft.network.RegistryFriendlyByteBuf
 import net.minecraft.network.chat.Component
 import net.minecraft.network.chat.MutableComponent
 import net.minecraft.network.codec.StreamCodec
@@ -53,13 +56,13 @@ import net.minecraft.world.level.ItemLike
 import net.neoforged.bus.api.IEventBus
 import net.neoforged.fml.ModList
 import net.neoforged.neoforge.capabilities.Capabilities
-import net.neoforged.neoforge.common.crafting.IShapedRecipe
 import net.neoforged.neoforge.event.AddReloadListenerEvent
 import net.neoforged.neoforge.event.OnDatapackSyncEvent
 import net.neoforged.neoforge.event.RegisterCommandsEvent
 import net.neoforged.neoforge.event.entity.player.ItemTooltipEvent
 import net.neoforged.neoforge.event.server.ServerStartedEvent
 import net.neoforged.neoforge.network.PacketDistributor
+import net.neoforged.neoforge.network.connection.ConnectionType
 import net.neoforged.neoforge.network.handling.IPayloadContext
 import net.neoforged.neoforge.registries.DeferredRegister
 import net.neoforged.neoforge.server.command.EnumArgument
@@ -86,6 +89,7 @@ import ru.dbotthepony.mc.otm.core.getReverseTag
 import ru.dbotthepony.mc.otm.core.isNotEmpty
 import ru.dbotthepony.mc.otm.core.math.Decimal
 import ru.dbotthepony.mc.otm.core.math.isZero
+import ru.dbotthepony.mc.otm.core.readComponent
 import ru.dbotthepony.mc.otm.core.readItemType
 import ru.dbotthepony.mc.otm.core.registryName
 import ru.dbotthepony.mc.otm.core.stream
@@ -97,6 +101,7 @@ import ru.dbotthepony.mc.otm.core.util.readBinaryComponent
 import ru.dbotthepony.mc.otm.core.util.readCollection
 import ru.dbotthepony.mc.otm.core.util.writeBinaryComponent
 import ru.dbotthepony.mc.otm.core.util.writeCollection
+import ru.dbotthepony.mc.otm.core.writeComponent
 import ru.dbotthepony.mc.otm.core.writeItemType
 import ru.dbotthepony.mc.otm.matter.MatterManager.Finder
 import ru.dbotthepony.mc.otm.milliTime
@@ -1706,7 +1711,7 @@ object MatterManager {
 				matterValues[item] = value
 		}
 
-		syncRegistry(PacketDistributor::sendToAllPlayers)
+		syncRegistry(server.registryAccess(), PacketDistributor::sendToAllPlayers)
 	}
 
 	fun onDataPackSync(event: OnDatapackSyncEvent) {
@@ -1714,9 +1719,9 @@ object MatterManager {
 			return
 
 		if (event.player == null) {
-			syncRegistry(PacketDistributor::sendToAllPlayers)
+			syncRegistry(event.playerList.server.registryAccess(), PacketDistributor::sendToAllPlayers)
 		} else {
-			syncRegistry { PacketDistributor.sendToPlayer(event.player as ServerPlayer, it) }
+			syncRegistry(event.playerList.server.registryAccess()) { PacketDistributor.sendToPlayer(event.player as ServerPlayer, it) }
 		}
 	}
 
@@ -1732,56 +1737,65 @@ object MatterManager {
 
 	private val receivedPackets = ArrayList<SyncPacket>()
 
-	private fun syncRegistry(distributor: (CustomPacketPayload) -> Unit) {
+	private fun syncRegistry(registry: RegistryAccess, distributor: (CustomPacketPayload) -> Unit) {
 		val time = SystemTime()
-		val stream = FastByteArrayOutputStream()
+		val writable = ByteBufAllocator.DEFAULT.heapBuffer()
-		val data = DataOutputStream(stream)
-
-		var commentsSize = commentary.size
-		data.writeInt(matterValues.size)
-
-		for ((k, v) in matterValues) {
-			data.writeItemType(k)
-			data.writeMatterValue(v)
-
-			val comment = commentary[k]
-
-			if (comment != null) {
-				commentsSize--
-				data.write(1)
-				data.writeCollection(comment, OutputStream::writeBinaryComponent)
-			} else {
-				data.write(0)
-			}
-		}
-
-		data.writeInt(commentsSize)
-
-		for ((k, v) in commentary) {
-			if (!matterValues.containsKey(k)) {
-				data.writeItemType(k)
-				data.writeCollection(v, OutputStream::writeBinaryComponent)
-			}
-		}
-
-		val deflater = Deflater(5)
-		deflater.setInput(stream.array, 0, stream.length)
-		deflater.finish()
 
 		val chunks = ArrayList<SyncPacket>()
-		var totalSize = 0
+		val totalSize: Int
-		var first = true
+		var compressedSize: Int = 0
 
-		while (true) {
+		try {
-			val bytes = ByteArray(2 shl 20 - 1024)
+			val data = RegistryFriendlyByteBuf(writable, registry, ConnectionType.NEOFORGE)
-			val written = deflater.deflate(bytes)
 
-			if (written == 0) {
+			var commentsSize = commentary.size
-				break
+			data.writeInt(matterValues.size)
-			} else {
+
-				totalSize += written
+			for ((k, v) in matterValues) {
-				chunks.add(SyncPacket(bytes, written, if (first) { first = false; FIRST } else NORMAL))
+				data.writeItemType(k)
+				data.writeMatterValue(v)
+
+				val comment = commentary[k]
+
+				if (comment != null) {
+					commentsSize--
+					data.writeBoolean(true)
+					data.writeCollection(comment) { _, it -> data.writeComponent(it) }
+				} else {
+					data.writeBoolean(false)
+				}
 			}
+
+			data.writeInt(commentsSize)
+
+			for ((k, v) in commentary) {
+				if (!matterValues.containsKey(k)) {
+					data.writeItemType(k)
+					data.writeCollection(v)  { _, it -> data.writeComponent(it) }
+				}
+			}
+
+			val deflater = Deflater(5)
+			deflater.setInput(data.array(), data.arrayOffset(), data.readableBytes())
+			deflater.finish()
+
+			totalSize = data.readableBytes()
+
+			var first = true
+
+			while (true) {
+				val bytes = ByteArray(2 shl 20 - 1024 /* 1 MiB - 1 KiB */)
+				val written = deflater.deflate(bytes)
+
+				if (written == 0) {
+					break
+				} else {
+					compressedSize += written
+					chunks.add(SyncPacket(bytes, written, if (first) { first = false; FIRST } else NORMAL))
+				}
+			}
+		} finally {
+			writable.release()
 		}
 
 		if (chunks.size == 1) {
@@ -1790,14 +1804,14 @@ object MatterManager {
 			chunks.last().mode = LAST
 		}
 
-		LOGGER.debug("Encoding matter registry packet took ${time.millis}ms, (${stream.length} bytes total, $totalSize bytes compressed)")
+		LOGGER.debug("Encoding matter registry packets took ${time.millis}ms, (${totalSize} bytes total, $compressedSize bytes compressed)")
 
 		for (chunk in chunks) {
 			distributor.invoke(chunk)
 		}
 	}
 
-	private fun playRegistryPackets() {
+	private fun playRegistryPackets(registry: RegistryAccess) {
 		val time = SystemTime()
 		var totalCompressedSize = 0
 
@@ -1820,66 +1834,53 @@ object MatterManager {
 
 		receivedPackets.clear()
 
-		val chunks = ArrayList<ByteArray>()
+		val spliced = ByteBufAllocator.DEFAULT.buffer(receivedPackets.sumOf { it.length })
-		var size = 0
-		val inflater = Inflater()
-		inflater.setInput(compressed)
 
-		while (!inflater.finished()) {
+		try {
-			val chunk = ByteArray(2 shl 16)
+			val buffer = ByteArray(2 shl 16)
-			val inflated = inflater.inflate(chunk)
+			val inflater = Inflater()
+			inflater.setInput(compressed)
 
-			if (inflated == 0) {
+			while (!inflater.finished()) {
-				break
+				val inflated = inflater.inflate(buffer)
-			} else {
-				size += inflated
 
-				if (size >= 1 shl 24 /* 16 MiB */) {
+				if (inflated == 0) {
-					throw IndexOutOfBoundsException("Pipe Bomb")
+					break
-				}
-
-				if (inflated == 2 shl 16) {
-					chunks.add(chunk)
 				} else {
-					chunks.add(chunk.copyOfRange(0, inflated))
+					spliced.writeBytes(buffer, 0, inflated)
+
+					if (spliced.writerIndex() >= 1 shl 24 /* 16 MiB */) {
+						throw IndexOutOfBoundsException("Pipe Bomb")
+					}
 				}
 			}
-		}
 
-		val spliced = ByteArray(size)
+			val data = RegistryFriendlyByteBuf(spliced, registry, ConnectionType.NEOFORGE)
-		var pointer2 = 0
+			val valuesSize = data.readInt()
 
-		for (chunk in chunks) {
+			matterValues.clear()
-			for (i in chunk.indices) {
+			commentary.clear()
-				spliced[pointer2++] = chunk[i]
+
+			for (i in 0 until valuesSize) {
+				val type = data.readItemType()
+				check(matterValues.put(type, data.readMatterValue()) == null) { "Duplicate item type $type" }
+
+				if (data.readBoolean()) {
+					commentary[type] = data.readCollection(::ArrayList) { data.readComponent() }
+				}
 			}
-		}
 
-		val stream = FastByteArrayInputStream(spliced)
+			val commentsSize = data.readInt()
-		val data = DataInputStream(stream)
 
-		val valuesSize = data.readInt()
+			for (i in 0 until commentsSize) {
-
+				val type = data.readItemType()
-		matterValues.clear()
+				check(commentary.put(type, data.readCollection(::ArrayList) { data.readComponent() }) == null) { "Duplicate commentary item type $type" }
-		commentary.clear()
-
-		for (i in 0 until valuesSize) {
-			val type = data.readItemType()
-			check(matterValues.put(type, data.readMatterValue()) == null) { "Duplicate item type $type" }
-
-			if (data.read() > 0) {
-				commentary[type] = data.readCollection { readBinaryComponent()!! }
 			}
+
+			LOGGER.debug("Decoding matter registry packets took ${time.millis}ms ($totalCompressedSize bytes compressed, ${spliced.writerIndex()} bytes total)")
+		} finally {
+			spliced.release()
 		}
-
-		val commentsSize = data.readInt()
-
-		for (i in 0 until commentsSize) {
-			val type = data.readItemType()
-			check(commentary.put(type, data.readCollection { readBinaryComponent()!! }) == null) { "Duplicate commentary item type $type" }
-		}
-
-		LOGGER.debug("Decoding matter registry packets took ${time.millis}ms ($totalCompressedSize bytes compressed, $size bytes total)")
 	}
 
 	fun readSyncPacket(buff: FriendlyByteBuf): SyncPacket {
@@ -1918,11 +1919,11 @@ object MatterManager {
 				receivedPackets.add(this)
 			} else if (mode == LAST) {
 				receivedPackets.add(this)
-				playRegistryPackets()
+				playRegistryPackets(context.player().registryAccess())
 			} else if (mode == FIRST_AND_LAST) {
 				receivedPackets.clear()
 				receivedPackets.add(this)
-				playRegistryPackets()
+				playRegistryPackets(context.player().registryAccess())
 			} else {
 				receivedPackets.add(this)
 			}