Add support for packet compression

blokk-api/src/main/kotlin/space/blokk/chat/ChatColor.kt

@@ -35,24 +35,26 @@ sealed class ChatColor(val stringRepresentation: String) {
     companion object {
         private val HEX_COLOR_REGEX = Regex("^#(?:[0-9a-fA-F]{3}){1,2}\$")
 
-        val NAMED_COLORS = setOf(
+        val NAMED_COLORS by lazy {
-            BLACK,
+            setOf(
-            DARK_BLUE,
+                BLACK,
-            DARK_GREEN,
+                DARK_BLUE,
-            DARK_AQUA,
+                DARK_GREEN,
-            DARK_RED,
+                DARK_AQUA,
-            PURPLE,
+                DARK_RED,
-            GOLD,
+                PURPLE,
-            GRAY,
+                GOLD,
-            DARK_GRAY,
+                GRAY,
-            BLUE,
+                DARK_GRAY,
-            GREEN,
+                BLUE,
-            AQUA,
+                GREEN,
-            RED,
+                AQUA,
-            PINK,
+                RED,
-            YELLOW,
+                PINK,
-            WHITE
+                YELLOW,
-        ).map { it.stringRepresentation to it }.toMap()
+                WHITE
+            ).map { it.stringRepresentation to it }.toMap()
+        }
 
         fun fromString(value: String): ChatColor =
             if (value.startsWith("#")) Hex(value)

blokk-server/src/main/kotlin/space/blokk/config/BlokkConfig.kt

@@ -8,5 +8,6 @@ data class BlokkConfig(
     val silentNonServerErrors: Boolean,
     val authenticateAndEncrypt: Boolean,
     val developmentMode: Boolean,
-    val minLogLevel: Logger.Level
+    val minLogLevel: Logger.Level,
+    val packetCompressionThreshold: Int
 )

blokk-server/src/main/kotlin/space/blokk/net/BlokkChannelInitializer.kt

@@ -5,14 +5,13 @@ import io.netty.channel.ChannelException
 import io.netty.channel.ChannelInitializer
 import io.netty.channel.ChannelOption
 import io.netty.handler.timeout.IdleStateHandler
-import java.util.concurrent.atomic.AtomicBoolean
 
 class BlokkChannelInitializer(private val blokkSocketServer: BlokkSocketServer) : ChannelInitializer<Channel>() {
     override fun initChannel(channel: Channel) {
-        if (!ipTOSFailed.get()) try {
+        if (!ipTOSFailed) try {
             channel.config().setOption(ChannelOption.IP_TOS, 0x18)
         } catch (e: ChannelException) {
-            ipTOSFailed.set(true)
+            ipTOSFailed = true
             blokkSocketServer.server.logger warn "Your OS does not support IP type of service"
         }
 
@@ -26,6 +25,7 @@ class BlokkChannelInitializer(private val blokkSocketServer: BlokkSocketServer)
     }
 
     companion object {
-        private var ipTOSFailed = AtomicBoolean(false)
+        @Volatile
+        private var ipTOSFailed = false
     }
 }

blokk-server/src/main/kotlin/space/blokk/net/BlokkSession.kt

@@ -101,16 +101,18 @@ class BlokkSession(private val channel: Channel, val server: BlokkServer) : Sess
     }
 
     fun enableEncryptionCodec(key: SecretKey) {
-        logger trace "Enabling encryption"
+        logger trace "Enabling encryption codec"
-
+        if (channel.pipeline().get("encryption") != null) return
-        if (channel.pipeline().get("encryption") != null) {
-            logger debug "Encryption already enabled"
-            return
-        }
-
         channel.pipeline().addBefore("framing", "encryption", EncryptionCodec(this, key))
     }
 
+    fun enableCompressionCodec() {
+        logger trace "Enabling compression codec"
+        if (channel.pipeline().get("compression") != null) return
+        channel.pipeline()
+            .addAfter("framing", "compression", CompressionCodec(server.config.packetCompressionThreshold))
+    }
+
     override suspend fun disconnect(reason: TextComponent, loggableReason: String) {
         fun getReason() =
             if (server.config.developmentMode) {

blokk-server/src/main/kotlin/space/blokk/net/CompressionCodec.kt

@@ -0,0 +1,51 @@
+package space.blokk.net
+
+import io.netty.buffer.ByteBuf
+import io.netty.buffer.Unpooled
+import io.netty.channel.ChannelHandlerContext
+import io.netty.handler.codec.ByteToMessageCodec
+import space.blokk.net.MinecraftProtocolDataTypes.readVarInt
+import space.blokk.net.MinecraftProtocolDataTypes.writeVarInt
+import java.util.zip.Deflater
+import java.util.zip.Inflater
+
+class CompressionCodec(private val threshold: Int) : ByteToMessageCodec<ByteBuf>() {
+    private val deflater = Deflater(3)
+    private val deflationBuffer = ByteArray(8192)
+    private val inflater = Inflater()
+
+    override fun encode(ctx: ChannelHandlerContext, msg: ByteBuf, out: ByteBuf) {
+        if (msg.readableBytes() >= threshold) {
+            out.writeVarInt(msg.readableBytes())
+
+            deflater.setInput(msg.nioBuffer())
+            deflater.finish()
+
+            while (!deflater.finished()) {
+                out.writeBytes(deflationBuffer, 0, deflater.deflate(deflationBuffer))
+            }
+
+            deflater.reset()
+        } else {
+            out.writeVarInt(0)
+            out.writeBytes(msg)
+        }
+    }
+
+    override fun decode(ctx: ChannelHandlerContext, msg: ByteBuf, out: MutableList<Any>) {
+        val length = msg.readVarInt()
+        val realLength = msg.readableBytes()
+
+        if (length == 0) {
+            out.add(msg.readRetainedSlice(realLength))
+        } else {
+            val buffer = Unpooled.buffer(length)
+
+            inflater.setInput(msg.nioBuffer())
+            inflater.inflate(buffer.nioBuffer())
+            inflater.reset()
+
+            out.add(buffer)
+        }
+    }
+}

blokk-server/src/main/kotlin/space/blokk/net/LoginAndJoinProcedure.kt

@@ -9,10 +9,7 @@ import space.blokk.event.ifCancelled
 import space.blokk.net.MinecraftProtocolDataTypes.writeString
 import space.blokk.net.event.PlayerInitializationEvent
 import space.blokk.net.event.SessionAfterLoginEvent
-import space.blokk.net.packet.login.EncryptionRequestPacket
+import space.blokk.net.packet.login.*
-import space.blokk.net.packet.login.EncryptionResponsePacket
-import space.blokk.net.packet.login.LoginStartPacket
-import space.blokk.net.packet.login.LoginSuccessPacket
 import space.blokk.net.packet.play.*
 import space.blokk.player.BlokkPlayer
 import space.blokk.player.GameMode
@@ -68,6 +65,9 @@ class LoginAndJoinProcedure(val session: BlokkSession) {
 
         val result = AuthenticationHelper.authenticate(hashString, state.username)
 
+        session.send(SetCompressionPacket(session.server.config.packetCompressionThreshold))
+        session.enableCompressionCodec()
+
         session.send(LoginSuccessPacket(result.uuid, result.username))
         session.state = Session.State.LoginSucceeded(result.username, result.uuid)
 

blokk-server/src/main/kotlin/space/blokk/util/Ticker.kt

@@ -3,19 +3,19 @@ package space.blokk.util
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ReceiveChannel
 import space.blokk.server.Server
-import java.util.*
+import java.util.concurrent.CopyOnWriteArraySet
 
 class Ticker {
     private val thread = Thread {
         val interval = 1000L / Server.TICKS_PER_SECOND
 
         while (true) {
-            channels.toSet().forEach { it.offer(Unit) }
+            channels.forEach { it.offer(Unit) }
             Thread.sleep(interval)
         }
     }
 
-    private val channels: MutableSet<Channel<Unit>> = Collections.synchronizedSet(mutableSetOf<Channel<Unit>>())
+    private val channels: MutableSet<Channel<Unit>> = CopyOnWriteArraySet()
 
     fun createTickChannel(): ReceiveChannel<Unit> {
         return Channel<Unit>().also { channel ->

blokk-server/src/main/resources/default-config.yml

@@ -4,3 +4,4 @@ silentNonServerErrors: true
 authenticateAndEncrypt: true
 developmentMode: false
 minLogLevel: INFO
+packetCompressionThreshold: 256