Pass compressed packets directly to the backend server/client

api/src/main/java/com/velocitypowered/api/proxy/config/ProxyConfig.java

@@ -114,6 +114,13 @@ public interface ProxyConfig {
    */
   int getCompressionLevel();
 
+  /**
+   * Get the minimum packet size required to check whether a packet can be passed as is.
+   *
+   * @return the decompression threshold
+   */
+  int getDecompressionThreshold();
+
   /**
    * Get the limit for how long a player must wait to log back in.
    *

native/src/main/java/com/velocitypowered/natives/compression/JavaVelocityCompressor.java

@@ -78,6 +78,33 @@ public void inflate(ByteBuf source, ByteBuf destination, int uncompressedSize)
     }
   }
 
+  @Override
+  public void inflatePartial(ByteBuf source, ByteBuf destination, int size)
+      throws DataFormatException {
+    ensureNotDisposed();
+
+    // We (probably) can't nicely deal with >=1 buffer nicely, so let's scream loudly.
+    checkArgument(source.nioBufferCount() == 1, "source has multiple backing buffers");
+    checkArgument(destination.nioBufferCount() == 1, "destination has multiple backing buffers");
+
+    final int origReaderIdx = source.readerIndex();
+
+    destination.ensureWritable(size);
+
+    try {
+      while (!inflater.finished() && size > 0) {
+        inflater.setInput(source.nioBuffer());
+        int produced = inflater.inflate(
+            destination.nioBuffer(destination.writerIndex(), size));
+        size -= produced;
+        destination.writerIndex(destination.writerIndex() + produced);
+        source.readerIndex(origReaderIdx + inflater.getTotalIn());
+      }
+    } finally {
+      inflater.reset();
+    }
+  }
+
   @Override
   public void deflate(ByteBuf source, ByteBuf destination) throws DataFormatException {
     ensureNotDisposed();

native/src/main/java/com/velocitypowered/natives/compression/VelocityCompressor.java

@@ -30,5 +30,10 @@ public interface VelocityCompressor extends Disposable, Native {
   void inflate(ByteBuf source, ByteBuf destination, int uncompressedSize)
       throws DataFormatException;
 
+  default void inflatePartial(ByteBuf source, ByteBuf destination, int size)
+      throws DataFormatException {
+    throw new UnsupportedOperationException();
+  }
+
   void deflate(ByteBuf source, ByteBuf destination) throws DataFormatException;
 }

proxy/src/main/java/com/velocitypowered/proxy/config/VelocityConfiguration.java

@@ -323,6 +323,11 @@ public int getCompressionLevel() {
     return advanced.getCompressionLevel();
   }
 
+  @Override
+  public int getDecompressionThreshold() {
+    return advanced.getDecompressionThreshold();
+  }
+
   @Override
   public int getLoginRatelimit() {
     return advanced.getLoginRatelimit();
@@ -737,6 +742,8 @@ private static class Advanced {
     @Expose
     private int compressionLevel = -1;
     @Expose
+    private int decompressionThreshold = 2048;
+    @Expose
     private int loginRatelimit = 3000;
     @Expose
     private int connectionTimeout = 5000;
@@ -766,6 +773,7 @@ private Advanced(CommentedConfig config) {
       if (config != null) {
         this.compressionThreshold = config.getIntOrElse("compression-threshold", 256);
         this.compressionLevel = config.getIntOrElse("compression-level", -1);
+        this.decompressionThreshold = config.getIntOrElse("decompression-threshold", 2048);
         this.loginRatelimit = config.getIntOrElse("login-ratelimit", 3000);
         this.connectionTimeout = config.getIntOrElse("connection-timeout", 5000);
         this.readTimeout = config.getIntOrElse("read-timeout", 30000);
@@ -793,6 +801,10 @@ public int getCompressionLevel() {
       return compressionLevel;
     }
 
+    public int getDecompressionThreshold() {
+      return decompressionThreshold;
+    }
+
     public int getLoginRatelimit() {
       return loginRatelimit;
     }
@@ -842,6 +854,7 @@ public String toString() {
       return "Advanced{"
           + "compressionThreshold=" + compressionThreshold
           + ", compressionLevel=" + compressionLevel
+          + ", decompressionThreshold=" + decompressionThreshold
           + ", loginRatelimit=" + loginRatelimit
           + ", connectionTimeout=" + connectionTimeout
           + ", readTimeout=" + readTimeout

proxy/src/main/java/com/velocitypowered/proxy/connection/MinecraftConnection.java

@@ -24,10 +24,11 @@
 import static com.velocitypowered.proxy.network.Connections.FRAME_DECODER;
 import static com.velocitypowered.proxy.network.Connections.FRAME_ENCODER;
 import static com.velocitypowered.proxy.network.Connections.MINECRAFT_DECODER;
-import static com.velocitypowered.proxy.network.Connections.MINECRAFT_ENCODER;
+import static com.velocitypowered.proxy.network.Connections.MINECRAFT_PRE_ENCODER;
 
 import com.google.common.base.Preconditions;
 import com.velocitypowered.api.network.ProtocolVersion;
+import com.velocitypowered.natives.compression.JavaVelocityCompressor;
 import com.velocitypowered.natives.compression.VelocityCompressor;
 import com.velocitypowered.natives.encryption.VelocityCipher;
 import com.velocitypowered.natives.encryption.VelocityCipherFactory;
@@ -42,14 +43,13 @@
 import com.velocitypowered.proxy.protocol.VelocityConnectionEvent;
 import com.velocitypowered.proxy.protocol.netty.MinecraftCipherDecoder;
 import com.velocitypowered.proxy.protocol.netty.MinecraftCipherEncoder;
-import com.velocitypowered.proxy.protocol.netty.MinecraftCompressDecoder;
+import com.velocitypowered.proxy.protocol.netty.MinecraftCompressAndIdDecoder;
 import com.velocitypowered.proxy.protocol.netty.MinecraftCompressorAndLengthEncoder;
 import com.velocitypowered.proxy.protocol.netty.MinecraftDecoder;
-import com.velocitypowered.proxy.protocol.netty.MinecraftEncoder;
+import com.velocitypowered.proxy.protocol.netty.MinecraftPreEncoder;
 import com.velocitypowered.proxy.protocol.netty.MinecraftVarintLengthEncoder;
 import com.velocitypowered.proxy.protocol.netty.PlayPacketQueueHandler;
 import com.velocitypowered.proxy.util.except.QuietDecoderException;
-import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandler;
@@ -155,8 +155,8 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
         HAProxyMessage proxyMessage = (HAProxyMessage) msg;
         this.remoteAddress = new InetSocketAddress(proxyMessage.sourceAddress(),
             proxyMessage.sourcePort());
-      } else if (msg instanceof ByteBuf) {
-        activeSessionHandler.handleUnknown((ByteBuf) msg);
+      } else {
+        activeSessionHandler.handleUnknown(msg);
       }
     } finally {
       ReferenceCountUtil.release(msg);
@@ -362,7 +362,7 @@ public void setState(StateRegistry state) {
     ensureInEventLoop();
 
     this.state = state;
-    this.channel.pipeline().get(MinecraftEncoder.class).setState(state);
+    this.channel.pipeline().get(MinecraftPreEncoder.class).setState(state);
     this.channel.pipeline().get(MinecraftDecoder.class).setState(state);
 
     if (state == StateRegistry.CONFIG) {
@@ -379,9 +379,9 @@ public void setState(StateRegistry state) {
    */
   public void addPlayPacketQueueHandler() {
     if (this.channel.pipeline().get(Connections.PLAY_PACKET_QUEUE) == null) {
-      this.channel.pipeline().addAfter(Connections.MINECRAFT_ENCODER, Connections.PLAY_PACKET_QUEUE,
+      this.channel.pipeline().addAfter(MINECRAFT_PRE_ENCODER, Connections.PLAY_PACKET_QUEUE,
            new PlayPacketQueueHandler(this.protocolVersion,
-                channel.pipeline().get(MinecraftEncoder.class).getDirection()));
+                channel.pipeline().get(MinecraftPreEncoder.class).getDirection()));
     }
   }
 
@@ -400,11 +400,11 @@ public void setProtocolVersion(ProtocolVersion protocolVersion) {
     boolean changed = this.protocolVersion != protocolVersion;
     this.protocolVersion = protocolVersion;
     if (protocolVersion != ProtocolVersion.LEGACY) {
-      this.channel.pipeline().get(MinecraftEncoder.class).setProtocolVersion(protocolVersion);
+      this.channel.pipeline().get(MinecraftPreEncoder.class).setProtocolVersion(protocolVersion);
       this.channel.pipeline().get(MinecraftDecoder.class).setProtocolVersion(protocolVersion);
     } else {
       // Legacy handshake handling
-      this.channel.pipeline().remove(MINECRAFT_ENCODER);
+      this.channel.pipeline().remove(MINECRAFT_PRE_ENCODER);
       this.channel.pipeline().remove(MINECRAFT_DECODER);
     }
 
@@ -496,16 +496,17 @@ public void setCompressionThreshold(int threshold) {
     ensureInEventLoop();
 
     if (threshold == -1) {
-      final ChannelHandler removedDecoder = channel.pipeline().remove(COMPRESSION_DECODER);
+      channel.pipeline().replace(COMPRESSION_DECODER, COMPRESSION_DECODER,
+          new MinecraftCompressAndIdDecoder(server));
       final ChannelHandler removedEncoder = channel.pipeline().remove(COMPRESSION_ENCODER);
 
-      if (removedDecoder != null && removedEncoder != null) {
+      if (removedEncoder != null) {
         channel.pipeline().addBefore(MINECRAFT_DECODER, FRAME_ENCODER,
             MinecraftVarintLengthEncoder.INSTANCE);
         channel.pipeline().fireUserEventTriggered(VelocityConnectionEvent.COMPRESSION_DISABLED);
       }
     } else {
-      MinecraftCompressDecoder decoder = (MinecraftCompressDecoder) channel.pipeline()
+      MinecraftCompressAndIdDecoder decoder = (MinecraftCompressAndIdDecoder) channel.pipeline()
           .get(COMPRESSION_DECODER);
       MinecraftCompressorAndLengthEncoder encoder =
           (MinecraftCompressorAndLengthEncoder) channel.pipeline().get(COMPRESSION_ENCODER);
@@ -517,11 +518,12 @@ public void setCompressionThreshold(int threshold) {
         VelocityCompressor compressor = Natives.compress.get().create(level);
 
         encoder = new MinecraftCompressorAndLengthEncoder(threshold, compressor);
-        decoder = new MinecraftCompressDecoder(threshold, compressor);
+        decoder = new MinecraftCompressAndIdDecoder(threshold, compressor,
+            JavaVelocityCompressor.FACTORY.create(1), server);
 
         channel.pipeline().remove(FRAME_ENCODER);
-        channel.pipeline().addBefore(MINECRAFT_DECODER, COMPRESSION_DECODER, decoder);
-        channel.pipeline().addBefore(MINECRAFT_ENCODER, COMPRESSION_ENCODER, encoder);
+        channel.pipeline().replace(COMPRESSION_DECODER, COMPRESSION_DECODER, decoder);
+        channel.pipeline().addBefore(MINECRAFT_PRE_ENCODER, COMPRESSION_ENCODER, encoder);
 
         channel.pipeline().fireUserEventTriggered(VelocityConnectionEvent.COMPRESSION_ENABLED);
       }

proxy/src/main/java/com/velocitypowered/proxy/connection/MinecraftSessionHandler.java

@@ -70,7 +70,6 @@
 import com.velocitypowered.proxy.protocol.packet.title.TitleSubtitlePacket;
 import com.velocitypowered.proxy.protocol.packet.title.TitleTextPacket;
 import com.velocitypowered.proxy.protocol.packet.title.TitleTimesPacket;
-import io.netty.buffer.ByteBuf;
 
 /**
  * Interface for dispatching received Minecraft packets.
@@ -85,7 +84,7 @@ default void handleGeneric(MinecraftPacket packet) {
 
   }
 
-  default void handleUnknown(ByteBuf buf) {
+  default void handleUnknown(Object obj) {
 
   }
 

proxy/src/main/java/com/velocitypowered/proxy/connection/backend/BackendPlaySessionHandler.java

@@ -57,7 +57,6 @@
 import com.velocitypowered.proxy.protocol.packet.chat.ComponentHolder;
 import com.velocitypowered.proxy.protocol.packet.config.StartUpdate;
 import com.velocitypowered.proxy.protocol.util.PluginMessageUtil;
-import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
@@ -334,8 +333,8 @@ public void handleGeneric(MinecraftPacket packet) {
   }
 
   @Override
-  public void handleUnknown(ByteBuf buf) {
-    playerConnection.delayedWrite(buf.retain());
+  public void handleUnknown(Object obj) {
+    playerConnection.delayedWrite(obj);
     if (++packetsFlushed >= MAXIMUM_PACKETS_TO_FLUSH) {
       playerConnection.flush();
       packetsFlushed = 0;

proxy/src/main/java/com/velocitypowered/proxy/connection/client/AuthSessionHandler.java

@@ -42,7 +42,6 @@
 import com.velocitypowered.proxy.protocol.packet.LoginAcknowledged;
 import com.velocitypowered.proxy.protocol.packet.ServerLoginSuccess;
 import com.velocitypowered.proxy.protocol.packet.SetCompression;
-import io.netty.buffer.ByteBuf;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.UUID;
@@ -251,7 +250,7 @@ private CompletableFuture<Void> connectToInitialServer(ConnectedPlayer player) {
   }
 
   @Override
-  public void handleUnknown(ByteBuf buf) {
+  public void handleUnknown(Object obj) {
     mcConnection.close(true);
   }
 

proxy/src/main/java/com/velocitypowered/proxy/connection/client/ClientConfigSessionHandler.java

@@ -155,7 +155,7 @@ public void handleGeneric(MinecraftPacket packet) {
   }
 
   @Override
-  public void handleUnknown(ByteBuf buf) {
+  public void handleUnknown(Object obj) {
     VelocityServerConnection serverConnection = player.getConnectedServer();
     if (serverConnection == null) {
       // No server connection yet, probably transitioning.
@@ -164,7 +164,7 @@ public void handleUnknown(ByteBuf buf) {
 
     MinecraftConnection smc = serverConnection.getConnection();
     if (smc != null && !smc.isClosed() && serverConnection.getPhase().consideredComplete()) {
-      smc.write(buf.retain());
+      smc.write(obj);
     }
   }
 

proxy/src/main/java/com/velocitypowered/proxy/connection/client/ClientPlaySessionHandler.java

@@ -69,7 +69,6 @@
 import com.velocitypowered.proxy.protocol.packet.title.GenericTitlePacket;
 import com.velocitypowered.proxy.protocol.util.PluginMessageUtil;
 import com.velocitypowered.proxy.util.CharacterUtil;
-import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import io.netty.util.ReferenceCountUtil;
@@ -420,7 +419,7 @@ public void handleGeneric(MinecraftPacket packet) {
   }
 
   @Override
-  public void handleUnknown(ByteBuf buf) {
+  public void handleUnknown(Object obj) {
     VelocityServerConnection serverConnection = player.getConnectedServer();
     if (serverConnection == null) {
       // No server connection yet, probably transitioning.
@@ -429,7 +428,7 @@ public void handleUnknown(ByteBuf buf) {
 
     MinecraftConnection smc = serverConnection.getConnection();
     if (smc != null && !smc.isClosed() && serverConnection.getPhase().consideredComplete()) {
-      smc.write(buf.retain());
+      smc.write(obj);
     }
   }
 

proxy/src/main/java/com/velocitypowered/proxy/connection/client/ConnectedPlayer.java

@@ -58,7 +58,7 @@
 import com.velocitypowered.proxy.connection.util.ConnectionRequestResults.Impl;
 import com.velocitypowered.proxy.connection.util.VelocityInboundConnection;
 import com.velocitypowered.proxy.protocol.StateRegistry;
-import com.velocitypowered.proxy.protocol.netty.MinecraftEncoder;
+import com.velocitypowered.proxy.protocol.netty.MinecraftPreEncoder;
 import com.velocitypowered.proxy.protocol.packet.ClientSettings;
 import com.velocitypowered.proxy.protocol.packet.Disconnect;
 import com.velocitypowered.proxy.protocol.packet.HeaderAndFooter;
@@ -1129,7 +1129,7 @@ public void switchToConfigState() {
     CompletableFuture.runAsync(() -> {
       connection.write(new StartUpdate());
       connection.getChannel().pipeline()
-          .get(MinecraftEncoder.class).setState(StateRegistry.CONFIG);
+          .get(MinecraftPreEncoder.class).setState(StateRegistry.CONFIG);
       // Make sure we don't send any play packets to the player after update start
       connection.addPlayPacketQueueHandler();
     }, connection.eventLoop()).exceptionally((ex) -> {

proxy/src/main/java/com/velocitypowered/proxy/connection/client/HandshakeSessionHandler.java

@@ -35,7 +35,6 @@
 import com.velocitypowered.proxy.protocol.packet.LegacyDisconnect;
 import com.velocitypowered.proxy.protocol.packet.LegacyHandshake;
 import com.velocitypowered.proxy.protocol.packet.LegacyPing;
-import io.netty.buffer.ByteBuf;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.Optional;
@@ -198,7 +197,7 @@ public void handleGeneric(MinecraftPacket packet) {
   }
 
   @Override
-  public void handleUnknown(ByteBuf buf) {
+  public void handleUnknown(Object obj) {
     // Unknown packet received. Better to close the connection.
     connection.close(true);
   }

proxy/src/main/java/com/velocitypowered/proxy/connection/client/InitialLoginSessionHandler.java

@@ -40,7 +40,6 @@
 import com.velocitypowered.proxy.protocol.packet.EncryptionResponse;
 import com.velocitypowered.proxy.protocol.packet.LoginPluginResponse;
 import com.velocitypowered.proxy.protocol.packet.ServerLogin;
-import io.netty.buffer.ByteBuf;
 import java.net.InetSocketAddress;
 import java.security.GeneralSecurityException;
 import java.security.KeyPair;
@@ -284,7 +283,7 @@ private EncryptionRequest generateEncryptionRequest() {
   }
 
   @Override
-  public void handleUnknown(ByteBuf buf) {
+  public void handleUnknown(Object obj) {
     mcConnection.close(true);
   }
 

proxy/src/main/java/com/velocitypowered/proxy/connection/client/StatusSessionHandler.java

@@ -28,7 +28,6 @@
 import com.velocitypowered.proxy.protocol.packet.StatusRequest;
 import com.velocitypowered.proxy.protocol.packet.StatusResponse;
 import com.velocitypowered.proxy.util.except.QuietRuntimeException;
-import io.netty.buffer.ByteBuf;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -109,7 +108,7 @@ public boolean handle(StatusRequest packet) {
   }
 
   @Override
-  public void handleUnknown(ByteBuf buf) {
+  public void handleUnknown(Object obj) {
     // what even is going on?
     connection.close(true);
   }