Implment simple backpressure support

In most cases this should only trigger on initial spawns and server
switches.

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

@@ -107,6 +107,13 @@ public class MinecraftConnection extends ChannelInboundHandlerAdapter {
         }
     }
 
+    @Override
+    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
+        if (sessionHandler != null) {
+            sessionHandler.writabilityChanged();
+        }
+    }
+
     public void write(Object msg) {
         if (channel.isActive()) {
             channel.writeAndFlush(msg, channel.voidPromise());

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

@@ -29,4 +29,8 @@ public interface MinecraftSessionHandler {
     default void exception(Throwable throwable) {
 
     }
+
+    default void writabilityChanged() {
+
+    }
 }

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

@@ -4,6 +4,7 @@ import com.velocitypowered.api.event.connection.DisconnectEvent;
 import com.velocitypowered.api.proxy.messages.ChannelSide;
 import com.velocitypowered.api.proxy.messages.MessageHandler;
 import com.velocitypowered.proxy.VelocityServer;
+import com.velocitypowered.proxy.connection.backend.VelocityServerConnection;
 import com.velocitypowered.proxy.protocol.MinecraftPacket;
 import com.velocitypowered.proxy.protocol.ProtocolConstants;
 import com.velocitypowered.proxy.protocol.packet.*;
@@ -151,6 +152,15 @@ public class ClientPlaySessionHandler implements MinecraftSessionHandler {
                 .build());
     }
 
+    @Override
+    public void writabilityChanged() {
+        VelocityServerConnection server = player.getConnectedServer();
+        if (server != null) {
+            boolean writable = player.getConnection().getChannel().isWritable();
+            server.getMinecraftConnection().getChannel().config().setAutoRead(writable);
+        }
+    }
+
     public void handleBackendJoinGame(JoinGame joinGame) {
         resetPingData(); // reset ping data;
         if (!spawned) {

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

@@ -16,11 +16,7 @@ import com.velocitypowered.proxy.protocol.netty.MinecraftVarintFrameDecoder;
 import com.velocitypowered.proxy.protocol.netty.MinecraftVarintLengthEncoder;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.bootstrap.ServerBootstrap;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.EventLoopGroup;
+import io.netty.channel.*;
 import io.netty.channel.epoll.Epoll;
 import io.netty.channel.epoll.EpollDatagramChannel;
 import io.netty.channel.epoll.EpollEventLoopGroup;
@@ -48,6 +44,8 @@ import java.util.concurrent.TimeUnit;
 import static com.velocitypowered.proxy.network.Connections.*;
 
 public final class ConnectionManager {
+    private static final WriteBufferWaterMark SERVER_WRITE_MARK = new WriteBufferWaterMark(1 << 16, 1 << 18);
+
     private static final Logger logger = LogManager.getLogger(ConnectionManager.class);
 
     private final Set<Channel> endpoints = new HashSet<>();
@@ -72,6 +70,7 @@ public final class ConnectionManager {
         final ServerBootstrap bootstrap = new ServerBootstrap()
                 .channel(this.transportType.serverSocketChannelClass)
                 .group(this.bossGroup, this.workerGroup)
+                .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, SERVER_WRITE_MARK)
                 .childHandler(new ChannelInitializer<Channel>() {
                     @Override
                     protected void initChannel(final Channel ch) {