Various Netty changes.

- Potentially fixed a reference count leak with plugin messages.
- Cleaned up plugin message handling.
- Optimized the pipeline for better throughput by eliminating copying
  in the varint encoder and reduced object churn elsewhere.

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

@@ -46,23 +46,22 @@ public class BackendPlaySessionHandler implements MinecraftSessionHandler {
         } else if (packet instanceof PluginMessage) {
             PluginMessage pm = (PluginMessage) packet;
             try {
-                PluginMessage newPacket = pm;
+                if (!canForwardPluginMessage(pm)) {
-                if (!canForwardPluginMessage(newPacket)) {
                     return;
                 }
 
-                if (newPacket.getChannel().equals("MC|Brand")) {
+                if (PluginMessageUtil.isMCBrand(pm)) {
-                    newPacket = PluginMessageUtil.rewriteMCBrand(pm);
+                    connection.getProxyPlayer().getConnection().write(PluginMessageUtil.rewriteMCBrand(pm));
+                    return;
                 }
 
-                if (newPacket == pm) {
+                // we'll decrement this twice: once when writing to the server, once just below this block,
-                    // we'll decrement this thrice: once when writing to the server, once just below this block,
                 // and once in the MinecraftConnection (since this is a slice)
                 pm.getData().retain();
-                }
+
-                connection.getProxyPlayer().getConnection().write(newPacket);
+                connection.getProxyPlayer().getConnection().write(pm);
             } finally {
-                ReferenceCountUtil.release(pm.getData());
+                pm.getData().release();
             }
         } else {
             // Just forward the packet on. We don't have anything to handle at this time.

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

@@ -9,12 +9,14 @@ import com.velocitypowered.proxy.data.scoreboard.Scoreboard;
 import com.velocitypowered.proxy.data.scoreboard.Team;
 import com.velocitypowered.proxy.protocol.MinecraftPacket;
 import com.velocitypowered.proxy.protocol.ProtocolConstants;
+import com.velocitypowered.proxy.protocol.ProtocolUtils;
 import com.velocitypowered.proxy.protocol.packet.*;
 import com.velocitypowered.proxy.connection.MinecraftSessionHandler;
 import com.velocitypowered.proxy.protocol.remap.EntityIdRemapper;
 import com.velocitypowered.proxy.protocol.util.PluginMessageUtil;
 import com.velocitypowered.proxy.util.ThrowableUtils;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import io.netty.channel.EventLoop;
 import io.netty.util.ReferenceCountUtil;
 import net.kyori.text.TextComponent;
@@ -38,7 +40,6 @@ public class ClientPlaySessionHandler implements MinecraftSessionHandler {
     private boolean spawned = false;
     private final List<UUID> serverBossBars = new ArrayList<>();
     private final Set<String> clientPluginMsgChannels = new HashSet<>();
-    private PluginMessage brandMessage;
     private int currentDimension;
     private Scoreboard serverScoreboard = new Scoreboard();
     private EntityIdRemapper idRemapper;
@@ -110,10 +111,6 @@ public class ClientPlaySessionHandler implements MinecraftSessionHandler {
             pingTask.cancel(false);
             pingTask = null;
         }
-
-        if (brandMessage != null) {
-            brandMessage.getData().release();
-        }
     }
 
     @Override
@@ -179,12 +176,6 @@ public class ClientPlaySessionHandler implements MinecraftSessionHandler {
                     PluginMessageUtil.constructChannelsPacket(channel, clientPluginMsgChannels));
         }
 
-        // Tell the server the client's brand
-        if (brandMessage != null) {
-            brandMessage.getData().retain();
-            player.getConnectedServer().getMinecraftConnection().delayedWrite(brandMessage);
-        }
-
         // Flush everything
         player.getConnection().flush();
         player.getConnectedServer().getMinecraftConnection().flush();
@@ -201,7 +192,6 @@ public class ClientPlaySessionHandler implements MinecraftSessionHandler {
     public void handleClientPluginMessage(PluginMessage packet) {
         logger.info("Got client plugin message packet {}", packet);
 
-        PluginMessage original = packet;
         try {
             if (packet.getChannel().equals("REGISTER") || packet.getChannel().equals("minecraft:register")) {
                 List<String> actuallyRegistered = new ArrayList<>();
@@ -230,27 +220,16 @@ public class ClientPlaySessionHandler implements MinecraftSessionHandler {
                 clientPluginMsgChannels.removeAll(channels);
             }
 
-            if (packet.getChannel().equals("MC|Brand") || packet.getChannel().equals("minecraft:brand")) {
+            if (PluginMessageUtil.isMCBrand(packet)) {
-                if (this.brandMessage != null) {
+                player.getConnectedServer().getMinecraftConnection().write(PluginMessageUtil.rewriteMCBrand(packet));
-                    // Rewrite this packet to indicate that Velocity is running. Hurrah!
-                    packet = PluginMessageUtil.rewriteMCBrand(packet);
-                    this.brandMessage = packet;
-                } else {
-                    // Already have the brand packet and don't need this one.
                 return;
             }
-            }
 
-            // No other special handling?
+            // We're going to forward on the original packet.
-            if (packet == original) {
-                // we'll decrement this thrice: once when writing to the server, once just below this block,
-                // and once in the MinecraftConnection (since this is a slice)
             packet.getData().retain();
-            }
-
             player.getConnectedServer().getMinecraftConnection().write(packet);
         } finally {
-            ReferenceCountUtil.release(original.getData());
+            ReferenceCountUtil.release(packet.getData());
         }
     }
 

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

@@ -78,10 +78,8 @@ public class ConnectedPlayer implements MinecraftConnectionAssociation {
         String error = ThrowableUtils.briefDescription(throwable);
         String userMessage;
         if (connectedServer != null && connectedServer.getServerInfo().equals(info)) {
-            logger.error("{}: exception occurred in connection to {}", this, info.getName(), throwable);
             userMessage = "Exception in server " + info.getName();
         } else {
-            logger.error("{}: unable to connect to server {}", this, info.getName(), throwable);
             userMessage = "Exception connecting to server " + info.getName();
         }
         handleConnectionException(info, TextComponent.builder()

proxy/src/main/java/com/velocitypowered/proxy/protocol/netty/MinecraftCompressDecoder.java

@@ -25,7 +25,7 @@ public class MinecraftCompressDecoder extends MessageToMessageDecoder<ByteBuf> {
         int uncompressedSize = ProtocolUtils.readVarInt(msg);
         if (uncompressedSize == 0) {
             // Strip the now-useless uncompressed size, this message is already uncompressed.
-            out.add(msg.slice().retain());
+            out.add(msg.retainedSlice());
             msg.skipBytes(msg.readableBytes());
             return;
         }

proxy/src/main/java/com/velocitypowered/proxy/protocol/netty/MinecraftDecoder.java

@@ -26,7 +26,7 @@ public class MinecraftDecoder extends MessageToMessageDecoder<ByteBuf> {
             return;
         }
 
-        ByteBuf slice = msg.slice().retain();
+        ByteBuf slice = msg.retainedSlice();
 
         int packetId = ProtocolUtils.readVarInt(msg);
         MinecraftPacket packet = this.protocolVersion.createPacket(packetId);

proxy/src/main/java/com/velocitypowered/proxy/protocol/netty/MinecraftVarintFrameDecoder.java

@@ -21,7 +21,6 @@ public class MinecraftVarintFrameDecoder extends ByteToMessageDecoder {
             return;
         }
 
-        out.add(in.slice(in.readerIndex(), packetLength).retain());
+        out.add(in.readRetainedSlice(packetLength));
-        in.skipBytes(packetLength);
     }
 }

proxy/src/main/java/com/velocitypowered/proxy/protocol/netty/MinecraftVarintLengthEncoder.java

@@ -5,17 +5,21 @@ import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.MessageToByteEncoder;
+import io.netty.handler.codec.MessageToMessageEncoder;
+
+import java.util.List;
 
 @ChannelHandler.Sharable
-public class MinecraftVarintLengthEncoder extends MessageToByteEncoder<ByteBuf> {
+public class MinecraftVarintLengthEncoder extends MessageToMessageEncoder<ByteBuf> {
     public static final MinecraftVarintLengthEncoder INSTANCE = new MinecraftVarintLengthEncoder();
 
     private MinecraftVarintLengthEncoder() { }
 
     @Override
-    protected void encode(ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception {
+    protected void encode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> list) throws Exception {
-        out.ensureWritable(msg.readableBytes() + 5);
+        ByteBuf lengthBuf = ctx.alloc().buffer(5); // the maximum size of a varint
-        ProtocolUtils.writeVarInt(out, msg.readableBytes());
+        ProtocolUtils.writeVarInt(lengthBuf, buf.readableBytes());
-        out.writeBytes(msg);
+        list.add(lengthBuf);
+        list.add(buf.retain());
     }
 }

proxy/src/main/java/com/velocitypowered/proxy/protocol/util/PluginMessageUtil.java

@@ -15,7 +15,13 @@ import java.util.List;
 public enum PluginMessageUtil {
     ;
 
+    public static boolean isMCBrand(PluginMessage message) {
+        Preconditions.checkNotNull(message, "message");
+        return message.getChannel().equals("MC|Brand") || message.getChannel().equals("minecraft:brand");
+    }
+
     public static List<String> getChannels(PluginMessage message) {
+        Preconditions.checkNotNull(message, "message");
         Preconditions.checkArgument(message.getChannel().equals("REGISTER") ||
                 message.getChannel().equals("UNREGISTER") ||
                 message.getChannel().equals("minecraft:register") ||
@@ -26,6 +32,9 @@ public enum PluginMessageUtil {
     }
 
     public static PluginMessage constructChannelsPacket(String channel, Collection<String> channels) {
+        Preconditions.checkNotNull(channel, "channel");
+        Preconditions.checkNotNull(channel, "channels");
+
         PluginMessage message = new PluginMessage();
         message.setChannel(channel);
 
@@ -41,6 +50,9 @@ public enum PluginMessageUtil {
     }
 
     public static PluginMessage rewriteMCBrand(PluginMessage message) {
+        Preconditions.checkNotNull(message, "message");
+        Preconditions.checkArgument(isMCBrand(message), "message is not a MC Brand plugin message");
+
         ByteBuf rewrittenBuf = Unpooled.buffer();
         String currentBrand = ProtocolUtils.readString(message.getData());
         ProtocolUtils.writeString(rewrittenBuf, currentBrand + " (Velocity)");