Add support for server-side backpressure

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

@@ -195,6 +195,8 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
   public void write(Object msg) {
     if (channel.isActive()) {
       channel.writeAndFlush(msg, channel.voidPromise());
+    } else {
+      ReferenceCountUtil.release(msg);
     }
   }
 
@@ -205,6 +207,8 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
   public void delayedWrite(Object msg) {
     if (channel.isActive()) {
       channel.write(msg, channel.voidPromise());
+    } else {
+      ReferenceCountUtil.release(msg);
     }
   }
 

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

@@ -28,6 +28,7 @@ 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;
 import io.netty.handler.timeout.ReadTimeoutException;
 import java.util.Collection;
 import org.apache.logging.log4j.LogManager;
@@ -284,4 +285,11 @@ public class BackendPlaySessionHandler implements MinecraftSessionHandler {
       }
     }
   }
+
+  @Override
+  public void writabilityChanged() {
+    Channel serverChan = serverConn.ensureConnected().getChannel();
+    boolean writable = serverChan.isWritable();
+    playerConnection.setAutoReading(writable);
+  }
 }

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

@@ -300,7 +300,7 @@ public class ClientPlaySessionHandler implements MinecraftSessionHandler {
     boolean writable = player.getConnection().getChannel().isWritable();
 
     if (!writable) {
-      // We might have packets queued for the server, so flush them now to free up memory.
+      // We might have packets queued from the server, so flush them now to free up memory.
       player.getConnection().flush();
     }