Preliminary compression support.

src/main/java/io/minimum/minecraft/velocity/protocol/StateRegistry.java

@@ -33,6 +33,7 @@ public enum StateRegistry {
             TO_CLIENT.register(0x00, Disconnect.class, Disconnect::new);
             // Encryption Success will follow once Mojang auth/encryption is done
             TO_CLIENT.register(0x02, ServerLoginSuccess.class, ServerLoginSuccess::new);
+            TO_CLIENT.register(0x03, SetCompression.class, SetCompression::new);
         }
     };
 
@@ -50,7 +51,7 @@ public enum StateRegistry {
             this.state = state;
         }
 
-        public void register(int id, Class<? extends MinecraftPacket> clazz, Supplier<? extends MinecraftPacket> packetSupplier) {
+        public <P extends MinecraftPacket> void register(int id, Class<P> clazz, Supplier<P> packetSupplier) {
             idsToSuppliers.put(id, packetSupplier);
             packetClassesToIds.put(clazz, id);
         }

src/main/java/io/minimum/minecraft/velocity/protocol/netty/MinecraftCompressDecoder.java

@@ -0,0 +1,51 @@
+package io.minimum.minecraft.velocity.protocol.netty;
+
+import com.google.common.base.Preconditions;
+import io.minimum.minecraft.velocity.protocol.ProtocolUtils;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageDecoder;
+
+import java.util.List;
+import java.util.zip.Inflater;
+
+public class MinecraftCompressDecoder extends MessageToMessageDecoder<ByteBuf> {
+    private static final int MAXIMUM_INITIAL_BUFFER_SIZE = 65536; // 64KiB
+
+    private final int threshold;
+
+    public MinecraftCompressDecoder(int threshold) {
+        this.threshold = threshold;
+    }
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
+        int uncompressedSize = ProtocolUtils.readVarInt(msg);
+        if (uncompressedSize == 0) {
+            // Strip the now-useless uncompressed size, this message is already uncompressed.
+            out.add(msg.slice().retain());
+            return;
+        }
+
+        ByteBuf uncompressed = ctx.alloc().buffer(Math.min(uncompressedSize, MAXIMUM_INITIAL_BUFFER_SIZE));
+        try {
+            byte[] compressed = new byte[msg.readableBytes()];
+            msg.readBytes(compressed);
+            Inflater inflater = new Inflater();
+            inflater.setInput(compressed);
+
+            byte[] decompressed = new byte[8192];
+            while (!inflater.finished()) {
+                int inflatedBytes = inflater.inflate(decompressed);
+                uncompressed.writeBytes(decompressed, 0, inflatedBytes);
+            }
+
+            Preconditions.checkState(uncompressedSize == uncompressed.readableBytes(), "Mismatched compression sizes");
+            out.add(uncompressed);
+        } catch (Exception e) {
+            // If something went wrong, rethrow the exception, but ensure we free our temporary buffer first.
+            uncompressed.release();
+            throw e;
+        }
+    }
+}

src/main/java/io/minimum/minecraft/velocity/protocol/netty/MinecraftCompressEncoder.java

@@ -0,0 +1,47 @@
+package io.minimum.minecraft.velocity.protocol.netty;
+
+import io.minimum.minecraft.velocity.protocol.ProtocolUtils;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+
+import java.util.zip.Deflater;
+
+public class MinecraftCompressEncoder extends MessageToByteEncoder<ByteBuf> {
+    private final int threshold;
+
+    public MinecraftCompressEncoder(int threshold) {
+        this.threshold = threshold;
+    }
+
+    @Override
+    protected void encode(ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception {
+        if (msg.readableBytes() <= threshold) {
+            System.out.println("not compressing packet of 0x" + msg.readableBytes() + " size");
+            // Under the threshold, there is nothing to do.
+            ProtocolUtils.writeVarInt(out, 0);
+            out.writeBytes(msg);
+            return;
+        }
+
+        System.out.println("compressing packet of 0x" + msg.readableBytes() + " size");
+        Deflater deflater = new Deflater();
+        byte[] buf = new byte[msg.readableBytes()];
+        msg.readBytes(buf);
+        deflater.setInput(buf);
+        deflater.finish();
+
+        ByteBuf compressedBuffer = ctx.alloc().buffer();
+        try {
+            byte[] deflated = new byte[8192];
+            while (!deflater.finished()) {
+                int bytes = deflater.deflate(deflated);
+                compressedBuffer.writeBytes(deflated, 0, bytes);
+            }
+            ProtocolUtils.writeVarInt(out, buf.length);
+            out.writeBytes(compressedBuffer);
+        } finally {
+            compressedBuffer.release();
+        }
+    }
+}

src/main/java/io/minimum/minecraft/velocity/protocol/netty/MinecraftPipelineUtils.java

@@ -21,4 +21,12 @@ public class MinecraftPipelineUtils {
         ch.pipeline().addLast("minecraft-decoder", new MinecraftDecoder(ProtocolConstants.Direction.TO_CLIENT));
         ch.pipeline().addLast("minecraft-encoder", new MinecraftEncoder(ProtocolConstants.Direction.TO_SERVER));
     }
+
+    public static void enableCompression(Channel ch, int threshold) {
+        MinecraftCompressEncoder encoder = new MinecraftCompressEncoder(threshold);
+        MinecraftCompressDecoder decoder = new MinecraftCompressDecoder(threshold);
+
+        ch.pipeline().addBefore("minecraft-decoder", "compress-decoder", decoder);
+        ch.pipeline().addBefore("minecraft-encoder", "compress-encoder", encoder);
+    }
 }

src/main/java/io/minimum/minecraft/velocity/protocol/packets/SetCompression.java

@@ -0,0 +1,41 @@
+package io.minimum.minecraft.velocity.protocol.packets;
+
+import io.minimum.minecraft.velocity.protocol.MinecraftPacket;
+import io.minimum.minecraft.velocity.protocol.ProtocolConstants;
+import io.minimum.minecraft.velocity.protocol.ProtocolUtils;
+import io.netty.buffer.ByteBuf;
+
+public class SetCompression implements MinecraftPacket {
+    private int threshold;
+
+    public SetCompression() {}
+
+    public SetCompression(int threshold) {
+        this.threshold = threshold;
+    }
+
+    public int getThreshold() {
+        return threshold;
+    }
+
+    public void setThreshold(int threshold) {
+        this.threshold = threshold;
+    }
+
+    @Override
+    public String toString() {
+        return "SetCompression{" +
+                "threshold=" + threshold +
+                '}';
+    }
+
+    @Override
+    public void decode(ByteBuf buf, ProtocolConstants.Direction direction, int protocolVersion) {
+        this.threshold = ProtocolUtils.readVarInt(buf);
+    }
+
+    @Override
+    public void encode(ByteBuf buf, ProtocolConstants.Direction direction, int protocolVersion) {
+        ProtocolUtils.writeVarInt(buf, threshold);
+    }
+}

src/main/java/io/minimum/minecraft/velocity/proxy/InboundMinecraftConnection.java

@@ -6,8 +6,10 @@ import io.minimum.minecraft.velocity.protocol.ProtocolConstants;
 import io.minimum.minecraft.velocity.protocol.StateRegistry;
 import io.minimum.minecraft.velocity.protocol.netty.MinecraftDecoder;
 import io.minimum.minecraft.velocity.protocol.netty.MinecraftEncoder;
+import io.minimum.minecraft.velocity.protocol.netty.MinecraftPipelineUtils;
 import io.minimum.minecraft.velocity.protocol.packets.Handshake;
 import io.minimum.minecraft.velocity.protocol.packets.ServerLoginSuccess;
+import io.minimum.minecraft.velocity.protocol.packets.SetCompression;
 import io.minimum.minecraft.velocity.proxy.handler.HandshakeSessionHandler;
 import io.minimum.minecraft.velocity.proxy.handler.LoginSessionHandler;
 import io.minimum.minecraft.velocity.proxy.handler.PlaySessionHandler;
@@ -122,4 +124,9 @@ public class InboundMinecraftConnection {
         sessionHandler = new PlaySessionHandler(player, connection);
         connection.connect();
     }
+
+    public void enableCompression() {
+        write(new SetCompression(256));
+        MinecraftPipelineUtils.enableCompression(channel, 256);
+    }
 }

src/main/java/io/minimum/minecraft/velocity/proxy/handler/LoginSessionHandler.java

@@ -5,6 +5,7 @@ import io.minimum.minecraft.velocity.data.ServerInfo;
 import io.minimum.minecraft.velocity.protocol.MinecraftPacket;
 import io.minimum.minecraft.velocity.protocol.packets.ServerLogin;
 import io.minimum.minecraft.velocity.protocol.packets.ServerLoginSuccess;
+import io.minimum.minecraft.velocity.protocol.packets.SetCompression;
 import io.minimum.minecraft.velocity.proxy.*;
 
 import java.net.InetSocketAddress;
@@ -22,7 +23,9 @@ public class LoginSessionHandler implements MinecraftSessionHandler {
     public void handle(MinecraftPacket packet) {
         Preconditions.checkArgument(packet instanceof ServerLogin, "Expected a ServerLogin packet, not " + packet.getClass().getName());
 
-        // TODO: Encryption and compression
+        // TODO: Encryption
+        connection.enableCompression();
+
         String username = ((ServerLogin) packet).getUsername();
         ServerLoginSuccess success = new ServerLoginSuccess();
         success.setUsername(username);