Revert "Revert "Merge branch 'feature/flush-consolidation' into dev/1.1.0""

This reverts commit e476ee815a.

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

@@ -3,12 +3,14 @@ package com.velocitypowered.proxy.connection.backend;
 import static com.velocitypowered.proxy.VelocityServer.GSON;
 import static com.velocitypowered.proxy.connection.forge.legacy.LegacyForgeConstants.HANDSHAKE_HOSTNAME_TOKEN;
 import static com.velocitypowered.proxy.network.Connections.FLOW_HANDLER;
+import static com.velocitypowered.proxy.network.Connections.FLUSH_CONSOLIDATION;
 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.HANDLER;
 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.READ_TIMEOUT;
+import static io.netty.handler.flush.FlushConsolidationHandler.DEFAULT_EXPLICIT_FLUSH_AFTER_FLUSHES;
 
 import com.google.common.base.Preconditions;
 import com.velocitypowered.api.network.ProtocolVersion;
@@ -37,6 +39,7 @@ import io.netty.channel.Channel;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelInitializer;
 import io.netty.handler.flow.FlowControlHandler;
+import io.netty.handler.flush.FlushConsolidationHandler;
 import io.netty.handler.timeout.ReadTimeoutHandler;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -81,6 +84,8 @@ public class VelocityServerConnection implements MinecraftConnectionAssociation,
           @Override
           protected void initChannel(Channel ch) throws Exception {
             ch.pipeline()
+                .addLast(FLUSH_CONSOLIDATION, new FlushConsolidationHandler(
+                    DEFAULT_EXPLICIT_FLUSH_AFTER_FLUSHES, true))
                 .addLast(READ_TIMEOUT,
                     new ReadTimeoutHandler(server.getConfiguration().getReadTimeout(),
                         TimeUnit.MILLISECONDS))

proxy/src/main/java/com/velocitypowered/proxy/network/Connections.java

@@ -15,6 +15,7 @@ public class Connections {
   public static final String MINECRAFT_DECODER = "minecraft-decoder";
   public static final String MINECRAFT_ENCODER = "minecraft-encoder";
   public static final String READ_TIMEOUT = "read-timeout";
+  public static final String FLUSH_CONSOLIDATION = "flush-consolidation";
 
   private Connections() {
     throw new AssertionError();

proxy/src/main/java/com/velocitypowered/proxy/network/ServerChannelInitializer.java

@@ -1,5 +1,6 @@
 package com.velocitypowered.proxy.network;
 
+import static com.velocitypowered.proxy.network.Connections.FLUSH_CONSOLIDATION;
 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.LEGACY_PING_DECODER;
@@ -7,12 +8,12 @@ import static com.velocitypowered.proxy.network.Connections.LEGACY_PING_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.READ_TIMEOUT;
+import static io.netty.handler.flush.FlushConsolidationHandler.DEFAULT_EXPLICIT_FLUSH_AFTER_FLUSHES;
 
 import com.velocitypowered.proxy.VelocityServer;
 import com.velocitypowered.proxy.connection.MinecraftConnection;
 import com.velocitypowered.proxy.connection.client.HandshakeSessionHandler;
 import com.velocitypowered.proxy.protocol.ProtocolUtils;
-import com.velocitypowered.proxy.protocol.StateRegistry;
 import com.velocitypowered.proxy.protocol.netty.LegacyPingDecoder;
 import com.velocitypowered.proxy.protocol.netty.LegacyPingEncoder;
 import com.velocitypowered.proxy.protocol.netty.MinecraftDecoder;
@@ -22,6 +23,7 @@ import com.velocitypowered.proxy.protocol.netty.MinecraftVarintLengthEncoder;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelInitializer;
 import io.netty.handler.codec.haproxy.HAProxyMessageDecoder;
+import io.netty.handler.flush.FlushConsolidationHandler;
 import io.netty.handler.timeout.ReadTimeoutHandler;
 import java.util.concurrent.TimeUnit;
 
@@ -37,6 +39,8 @@ public class ServerChannelInitializer extends ChannelInitializer<Channel> {
   @Override
   protected void initChannel(final Channel ch) {
     ch.pipeline()
+        .addLast(FLUSH_CONSOLIDATION, new FlushConsolidationHandler(
+            DEFAULT_EXPLICIT_FLUSH_AFTER_FLUSHES, true))
         .addLast(READ_TIMEOUT,
             new ReadTimeoutHandler(this.server.getConfiguration().getReadTimeout(),
                 TimeUnit.MILLISECONDS))