Refactor of connection handling so we can share more logic.

src/main/java/io/minimum/minecraft/velocity/protocol/StateRegistry.java

@@ -24,9 +24,11 @@ public enum StateRegistry {
     PLAY {
         {
             TO_SERVER.register(0x02, Chat.class, Chat::new);
+            TO_SERVER.register(0x0b, Ping.class, Ping::new);
 
             TO_CLIENT.register(0x0F, Chat.class, Chat::new);
             TO_CLIENT.register(0x1A, Disconnect.class, Disconnect::new);
+            TO_CLIENT.register(0x1F, Ping.class, Ping::new);
         }
     },
     LOGIN {

src/main/java/io/minimum/minecraft/velocity/protocol/netty/MinecraftDecoder.java

@@ -3,7 +3,6 @@ package io.minimum.minecraft.velocity.protocol.netty;
 import com.google.common.base.Preconditions;
 import io.minimum.minecraft.velocity.protocol.*;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufUtil;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.CorruptedFrameException;
 import io.netty.handler.codec.MessageToMessageDecoder;
@@ -31,7 +30,6 @@ public class MinecraftDecoder extends MessageToMessageDecoder<ByteBuf> {
         int packetId = ProtocolUtils.readVarInt(msg);
         StateRegistry.ProtocolMappings mappings = direction == ProtocolConstants.Direction.TO_CLIENT ? state.TO_CLIENT : state.TO_SERVER;
         MinecraftPacket packet = mappings.createPacket(packetId);
-        //System.out.println(direction + " <- " + ByteBufUtil.hexDump(slice));
         if (packet == null) {
             msg.skipBytes(msg.readableBytes());
             out.add(new PacketWrapper(null, slice));
@@ -42,6 +40,7 @@ public class MinecraftDecoder extends MessageToMessageDecoder<ByteBuf> {
                 throw new CorruptedFrameException("Error decoding " + packet.getClass() + " Direction " + direction
                         + " Protocol " + protocolVersion + " State " + state + " ID " + Integer.toHexString(packetId), e);
             }
+            System.out.println("IN: " + packet);
             out.add(new PacketWrapper(packet, slice));
         }
     }

src/main/java/io/minimum/minecraft/velocity/protocol/netty/MinecraftEncoder.java

@@ -3,7 +3,6 @@ package io.minimum.minecraft.velocity.protocol.netty;
 import com.google.common.base.Preconditions;
 import io.minimum.minecraft.velocity.protocol.*;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufUtil;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.MessageToByteEncoder;
 
@@ -23,8 +22,7 @@ public class MinecraftEncoder extends MessageToByteEncoder<MinecraftPacket> {
         int packetId = mappings.getId(msg);
         ProtocolUtils.writeVarInt(out, packetId);
         msg.encode(out, direction, protocolVersion);
-
+        System.out.println("OUT: " + msg);
-        //System.out.println(direction + " -> " + ByteBufUtil.hexDump(out));
     }
 
     public int getProtocolVersion() {

src/main/java/io/minimum/minecraft/velocity/protocol/netty/MinecraftPipelineUtils.java

@@ -1,11 +1,14 @@
 package io.minimum.minecraft.velocity.protocol.netty;
 
 import io.minimum.minecraft.velocity.protocol.ProtocolConstants;
-import io.minimum.minecraft.velocity.protocol.compression.JavaVelocityCompressor;
 import io.netty.channel.Channel;
+import io.netty.handler.timeout.ReadTimeoutHandler;
+
+import java.util.concurrent.TimeUnit;
 
 public class MinecraftPipelineUtils {
     public static void strapPipelineForServer(Channel ch) {
+        ch.pipeline().addLast("read-timeout", new ReadTimeoutHandler(30, TimeUnit.SECONDS));
         ch.pipeline().addLast("legacy-ping-decode", new LegacyPingDecoder());
         ch.pipeline().addLast("frame-decoder", new MinecraftVarintFrameDecoder());
         ch.pipeline().addLast("legacy-ping-encode", LegacyPingEncoder.INSTANCE);
@@ -15,6 +18,7 @@ public class MinecraftPipelineUtils {
     }
 
     public static void strapPipelineForProxy(Channel ch) {
+        ch.pipeline().addLast("read-timeout", new ReadTimeoutHandler(30, TimeUnit.SECONDS));
         ch.pipeline().addLast("legacy-ping-decode", new LegacyPingDecoder());
         ch.pipeline().addLast("frame-decoder", new MinecraftVarintFrameDecoder());
         ch.pipeline().addLast("legacy-ping-encode", LegacyPingEncoder.INSTANCE);
@@ -22,19 +26,4 @@ public class MinecraftPipelineUtils {
         ch.pipeline().addLast("minecraft-decoder", new MinecraftDecoder(ProtocolConstants.Direction.TO_CLIENT));
         ch.pipeline().addLast("minecraft-encoder", new MinecraftEncoder(ProtocolConstants.Direction.TO_SERVER));
     }
-
-    public static void enableCompression(Channel ch, int threshold) {
-        if (threshold == -1) {
-            ch.pipeline().remove("compress-decoder");
-            ch.pipeline().remove("compress-encoder");
-            return;
-        }
-
-        JavaVelocityCompressor compressor = new JavaVelocityCompressor();
-        MinecraftCompressEncoder encoder = new MinecraftCompressEncoder(threshold, compressor);
-        MinecraftCompressDecoder decoder = new MinecraftCompressDecoder(threshold, compressor);
-
-        ch.pipeline().addBefore("minecraft-decoder", "compress-decoder", decoder);
-        ch.pipeline().addBefore("minecraft-encoder", "compress-encoder", encoder);
-    }
 }

src/main/java/io/minimum/minecraft/velocity/proxy/InboundMinecraftConnection.java

@@ -1,132 +0,0 @@
-package io.minimum.minecraft.velocity.proxy;
-
-import com.google.common.base.Preconditions;
-import io.minimum.minecraft.velocity.data.ServerInfo;
-import io.minimum.minecraft.velocity.protocol.ProtocolConstants;
-import io.minimum.minecraft.velocity.protocol.StateRegistry;
-import io.minimum.minecraft.velocity.protocol.netty.MinecraftDecoder;
-import io.minimum.minecraft.velocity.protocol.netty.MinecraftEncoder;
-import io.minimum.minecraft.velocity.protocol.netty.MinecraftPipelineUtils;
-import io.minimum.minecraft.velocity.protocol.packets.Handshake;
-import io.minimum.minecraft.velocity.protocol.packets.ServerLoginSuccess;
-import io.minimum.minecraft.velocity.protocol.packets.SetCompression;
-import io.minimum.minecraft.velocity.proxy.handler.HandshakeSessionHandler;
-import io.minimum.minecraft.velocity.proxy.handler.LoginSessionHandler;
-import io.minimum.minecraft.velocity.proxy.handler.PlaySessionHandler;
-import io.minimum.minecraft.velocity.proxy.handler.StatusSessionHandler;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
-import io.netty.util.AttributeKey;
-import net.kyori.text.TextComponent;
-
-import java.net.InetSocketAddress;
-import java.util.Optional;
-
-public class InboundMinecraftConnection {
-    public static final AttributeKey<InboundMinecraftConnection> CONNECTION = AttributeKey.newInstance("velocity-connection");
-
-    private final Channel channel;
-    private boolean closed;
-    private Handshake handshake;
-    private StateRegistry state;
-    private MinecraftSessionHandler sessionHandler;
-    private ConnectedPlayer connectedPlayer;
-
-    public InboundMinecraftConnection(Channel channel) {
-        this.channel = channel;
-        this.closed = false;
-        this.state = StateRegistry.HANDSHAKE;
-        this.sessionHandler = new HandshakeSessionHandler(this);
-    }
-
-    public void write(Object msg) {
-        ensureOpen();
-        channel.writeAndFlush(msg, channel.voidPromise());
-    }
-
-    public void closeWith(Object msg) {
-        ensureOpen();
-        teardown();
-        channel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
-            @Override
-            public void operationComplete(ChannelFuture future) throws Exception {
-                future.channel().close();
-            }
-        });
-    }
-
-    public void close() {
-        ensureOpen();
-        teardown();
-        channel.close();
-    }
-
-    public MinecraftSessionHandler getSessionHandler() {
-        return sessionHandler;
-    }
-
-    public void handleHandshake(Handshake handshake) {
-        ensureOpen();
-        Preconditions.checkNotNull(handshake, "handshake");
-        Preconditions.checkState(this.handshake == null, "Already handled a handshake for this connection!");
-        this.handshake = handshake;
-        switch (handshake.getNextStatus()) {
-            case 1:
-                // Status protocol
-                this.setStatus(StateRegistry.STATUS);
-                this.sessionHandler = new StatusSessionHandler(this);
-                break;
-            case 2:
-                this.setStatus(StateRegistry.LOGIN);
-                this.sessionHandler = new LoginSessionHandler(this);
-                break;
-            default:
-                throw new IllegalArgumentException("Invalid state " + handshake.getNextStatus());
-        }
-    }
-
-    private void ensureOpen() {
-        Preconditions.checkState(!closed, "Connection is closed.");
-    }
-
-    private void setStatus(StateRegistry state) {
-        Preconditions.checkNotNull(state, "state");
-        this.state = state;
-        channel.pipeline().get(MinecraftEncoder.class).setState(state);
-        channel.pipeline().get(MinecraftDecoder.class).setState(state);
-    }
-
-    public void teardown() {
-        closed = true;
-        if (connectedPlayer != null && connectedPlayer.getConnectedServer() != null) {
-            connectedPlayer.getConnectedServer().disconnect();
-        }
-    }
-
-    public boolean isClosed() {
-        return closed;
-    }
-
-    public Optional<ConnectedPlayer> getConnectedPlayer() {
-        return Optional.ofNullable(connectedPlayer);
-    }
-
-    public int getProtocolVersion() {
-        return handshake == null ? ProtocolConstants.MINECRAFT_1_12 : handshake.getProtocolVersion();
-    }
-
-    public void initiatePlay(ServerLoginSuccess success) {
-        setStatus(StateRegistry.PLAY);
-        ConnectedPlayer player = new ConnectedPlayer(success.getUsername(), success.getUuid(), this);
-        ServerInfo info = new ServerInfo("test", new InetSocketAddress("127.0.0.1", 25565));
-        ServerConnection connection = new ServerConnection(info, player, VelocityServer.getServer());
-        sessionHandler = new PlaySessionHandler(player, connection);
-        connection.connect();
-    }
-
-    public void enableCompression() {
-        write(new SetCompression(256));
-        MinecraftPipelineUtils.enableCompression(channel, 256);
-    }
-}

src/main/java/io/minimum/minecraft/velocity/proxy/MinecraftClientSessionHandler.java

@@ -1,51 +0,0 @@
-package io.minimum.minecraft.velocity.proxy;
-
-import io.minimum.minecraft.velocity.data.ServerPing;
-import io.minimum.minecraft.velocity.protocol.PacketWrapper;
-import io.minimum.minecraft.velocity.protocol.packets.*;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import net.kyori.text.TextComponent;
-
-public class MinecraftClientSessionHandler extends ChannelInboundHandlerAdapter {
-    @Override
-    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        InboundMinecraftConnection connection = ctx.channel().attr(InboundMinecraftConnection.CONNECTION).get();
-        if (msg instanceof PacketWrapper) {
-            PacketWrapper pw = (PacketWrapper) msg;
-            try {
-                if (pw.getPacket() == null) {
-                    connection.getSessionHandler().handleUnknown(pw.getBuffer());
-                } else {
-                    connection.getSessionHandler().handle(pw.getPacket());
-                }
-            } finally {
-                ((PacketWrapper) msg).getBuffer().release();
-            }
-        }
-
-        if (msg instanceof LegacyPing) {
-            // TODO: port this
-            System.out.println("Got LEGACY status request!");
-            ServerPing ping = new ServerPing(
-                    new ServerPing.Version(340, "1.12"),
-                    new ServerPing.Players(0, 0),
-                    TextComponent.of("this is a test"),
-                    null
-            );
-            LegacyPingResponse response = LegacyPingResponse.from(ping);
-            ctx.writeAndFlush(response);
-        }
-    }
-
-    @Override
-    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-        InboundMinecraftConnection connection = ctx.channel().attr(InboundMinecraftConnection.CONNECTION).get();
-        connection.teardown();
-    }
-
-    @Override
-    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-        cause.printStackTrace();
-    }
-}

src/main/java/io/minimum/minecraft/velocity/proxy/MinecraftConnection.java

@@ -0,0 +1,158 @@
+package io.minimum.minecraft.velocity.proxy;
+
+import com.google.common.base.Preconditions;
+import io.minimum.minecraft.velocity.protocol.PacketWrapper;
+import io.minimum.minecraft.velocity.protocol.StateRegistry;
+import io.minimum.minecraft.velocity.protocol.compression.JavaVelocityCompressor;
+import io.minimum.minecraft.velocity.protocol.netty.MinecraftCompressDecoder;
+import io.minimum.minecraft.velocity.protocol.netty.MinecraftCompressEncoder;
+import io.minimum.minecraft.velocity.protocol.netty.MinecraftDecoder;
+import io.minimum.minecraft.velocity.protocol.netty.MinecraftEncoder;
+import io.minimum.minecraft.velocity.protocol.packets.SetCompression;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.util.ReferenceCountUtil;
+
+/**
+ * A utility class to make working with the pipeline a little less painful and transparently handles certain Minecraft
+ * protocol mechanics.
+ */
+public class MinecraftConnection extends ChannelInboundHandlerAdapter {
+    private final Channel channel;
+    private boolean closed;
+    private StateRegistry state;
+    private MinecraftSessionHandler sessionHandler;
+    private int protocolVersion;
+
+    public MinecraftConnection(Channel channel) {
+        this.channel = channel;
+        this.closed = false;
+        this.state = StateRegistry.HANDSHAKE;
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        if (sessionHandler != null) {
+            sessionHandler.connected();
+        }
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        if (sessionHandler != null) {
+            sessionHandler.disconnected();
+        }
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        if (msg instanceof PacketWrapper) {
+            PacketWrapper pw = (PacketWrapper) msg;
+            try {
+                if (sessionHandler != null) {
+                    if (pw.getPacket() == null) {
+                        sessionHandler.handleUnknown(pw.getBuffer());
+                    } else {
+                        sessionHandler.handle(pw.getPacket());
+                    }
+                }
+            } finally {
+                ReferenceCountUtil.release(pw.getBuffer());
+            }
+        }
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        if (ctx.channel().isActive()) {
+            cause.printStackTrace();
+
+            if (sessionHandler != null) {
+                sessionHandler.exception(cause);
+            }
+
+            closed = true;
+            ctx.close();
+        }
+    }
+
+    public void write(Object msg) {
+        ensureOpen();
+        channel.writeAndFlush(msg, channel.voidPromise());
+    }
+
+    public void closeWith(Object msg) {
+        ensureOpen();
+        teardown();
+        channel.writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE);
+    }
+
+    public void close() {
+        ensureOpen();
+        teardown();
+        channel.close();
+    }
+
+    public void teardown() {
+        closed = true;
+    }
+
+    public Channel getChannel() {
+        return channel;
+    }
+
+    public boolean isClosed() {
+        return closed;
+    }
+
+    public StateRegistry getState() {
+        return state;
+    }
+
+    public void setState(StateRegistry state) {
+        this.state = state;
+        this.channel.pipeline().get(MinecraftEncoder.class).setState(state);
+        this.channel.pipeline().get(MinecraftDecoder.class).setState(state);
+    }
+
+    public int getProtocolVersion() {
+        return protocolVersion;
+    }
+
+    public void setProtocolVersion(int protocolVersion) {
+        this.protocolVersion = protocolVersion;
+        this.channel.pipeline().get(MinecraftEncoder.class).setProtocolVersion(protocolVersion);
+        this.channel.pipeline().get(MinecraftDecoder.class).setProtocolVersion(protocolVersion);
+    }
+
+    public MinecraftSessionHandler getSessionHandler() {
+        return sessionHandler;
+    }
+
+    public void setSessionHandler(MinecraftSessionHandler sessionHandler) {
+        this.sessionHandler = sessionHandler;
+    }
+
+    private void ensureOpen() {
+        Preconditions.checkState(!closed, "Connection is closed.");
+    }
+
+    public void setCompressionThreshold(int threshold) {
+        channel.writeAndFlush(new SetCompression(threshold), channel.voidPromise());
+
+        if (threshold == -1) {
+            channel.pipeline().remove("compress-decoder");
+            channel.pipeline().remove("compress-encoder");
+            return;
+        }
+
+        JavaVelocityCompressor compressor = new JavaVelocityCompressor();
+        MinecraftCompressEncoder encoder = new MinecraftCompressEncoder(threshold, compressor);
+        MinecraftCompressDecoder decoder = new MinecraftCompressDecoder(threshold, compressor);
+
+        channel.pipeline().addBefore("minecraft-decoder", "compress-decoder", decoder);
+        channel.pipeline().addBefore("minecraft-encoder", "compress-encoder", encoder);
+    }
+}

src/main/java/io/minimum/minecraft/velocity/proxy/MinecraftSessionHandler.java

@@ -10,7 +10,15 @@ public interface MinecraftSessionHandler {
         // No-op: we'll release the buffer later.
     }
 
-    default void connectionClosed() {
+    default void connected() {
+
+    }
+
+    default void disconnected() {
+
+    }
+
+    default void exception(Throwable throwable) {
 
     }
 }

src/main/java/io/minimum/minecraft/velocity/proxy/ServerConnection.java

@@ -1,143 +0,0 @@
-package io.minimum.minecraft.velocity.proxy;
-
-import io.minimum.minecraft.velocity.protocol.MinecraftPacket;
-import io.minimum.minecraft.velocity.protocol.PacketWrapper;
-import io.minimum.minecraft.velocity.protocol.StateRegistry;
-import io.minimum.minecraft.velocity.data.ServerInfo;
-import io.minimum.minecraft.velocity.protocol.netty.MinecraftDecoder;
-import io.minimum.minecraft.velocity.protocol.netty.MinecraftEncoder;
-import io.minimum.minecraft.velocity.protocol.netty.MinecraftPipelineUtils;
-import io.minimum.minecraft.velocity.protocol.packets.*;
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.*;
-import net.kyori.text.TextComponent;
-
-public class ServerConnection {
-    private Channel channel;
-    private final ServerInfo info;
-    private final ConnectedPlayer proxyPlayer;
-    private StateRegistry registry;
-    private final VelocityServer server;
-
-    public ServerConnection(ServerInfo target, ConnectedPlayer proxyPlayer, VelocityServer server) {
-        this.info = target;
-        this.proxyPlayer = proxyPlayer;
-        this.server = server;
-        this.registry = StateRegistry.HANDSHAKE;
-    }
-
-    public void connect() {
-        server.initializeGenericBootstrap()
-                .handler(new ChannelInitializer<Channel>() {
-                    @Override
-                    protected void initChannel(Channel ch) throws Exception {
-                        MinecraftPipelineUtils.strapPipelineForProxy(ch);
-                        ch.pipeline().addLast("state-based-interceptor", new StateBasedInterceptor());
-                    }
-                })
-                .connect(info.getAddress())
-                .addListener(new ChannelFutureListener() {
-                    @Override
-                    public void operationComplete(ChannelFuture future) throws Exception {
-                        if (future.isSuccess()) {
-                            channel = future.channel();
-                        } else {
-                            proxyPlayer.handleConnectionException(info, future.cause());
-                        }
-                    }
-                });
-    }
-
-    public void disconnect() {
-        channel.close();
-        channel = null;
-    }
-
-    public void forward(Object o) {
-        if (registry != StateRegistry.PLAY) {
-            throw new IllegalStateException("Not accepting player information until PLAY state");
-        }
-        channel.writeAndFlush(o, channel.voidPromise());
-    }
-
-    private class StateBasedInterceptor extends ChannelInboundHandlerAdapter {
-        @Override
-        public void channelActive(ChannelHandlerContext ctx) throws Exception {
-            // Initiate a handshake.
-            Handshake handshake = new Handshake();
-            handshake.setNextStatus(2); // login
-            handshake.setProtocolVersion(proxyPlayer.getConnection().getProtocolVersion());
-            handshake.setServerAddress(info.getAddress().getHostString());
-            handshake.setPort(info.getAddress().getPort());
-            ctx.writeAndFlush(handshake, ctx.voidPromise());
-
-            setRegistry(StateRegistry.LOGIN);
-
-            // Login
-            ServerLogin login = new ServerLogin();
-            login.setUsername(proxyPlayer.getUsername());
-            ctx.writeAndFlush(login, ctx.voidPromise());
-        }
-
-        @Override
-        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-            if (proxyPlayer.getConnection().isClosed()) {
-                // The upstream connection is closed, but we didn't forward that on for some reason. Close the connection
-                // here.
-                ctx.close();
-                return;
-            }
-
-            if (msg instanceof PacketWrapper) {
-                PacketWrapper pw = (PacketWrapper) msg;
-                try {
-                    switch (registry) {
-                        case LOGIN:
-                            onLogin(ctx, pw);
-                            break;
-                        case PLAY:
-                            onPlay(ctx, pw);
-                            break;
-                        default:
-                            throw new UnsupportedOperationException("Unsupported state " + registry);
-                    }
-                } finally {
-                    ((PacketWrapper) msg).getBuffer().release();
-                }
-            }
-        }
-
-        private void onPlay(ChannelHandlerContext ctx, PacketWrapper pw) {
-            proxyPlayer.getConnection().write(pw.getBuffer().retain());
-        }
-
-        private void onLogin(ChannelHandlerContext ctx, PacketWrapper wrapper) {
-            //System.out.println("FROM PROXIED SERVER -> " + wrapper.getPacket() + " / " + ByteBufUtil.hexDump(wrapper.getBuffer()));
-            MinecraftPacket packet = wrapper.getPacket();
-            if (packet instanceof Disconnect) {
-                Disconnect disconnect = (Disconnect) packet;
-                ctx.close();
-                proxyPlayer.handleConnectionException(info, disconnect);
-            }
-
-            if (packet instanceof SetCompression) {
-                System.out.println("Enabling compression on server connection, this is inefficient!");
-                SetCompression sc = (SetCompression) packet;
-                MinecraftPipelineUtils.enableCompression(channel, sc.getThreshold());
-            }
-
-            if (packet instanceof ServerLoginSuccess) {
-                // the player has been logged on.
-                System.out.println("Player connected to remote server");
-                setRegistry(StateRegistry.PLAY);
-                proxyPlayer.setConnectedServer(ServerConnection.this);
-            }
-        }
-    }
-
-    private void setRegistry(StateRegistry registry) {
-        this.registry = registry;
-        this.channel.pipeline().get(MinecraftEncoder.class).setState(registry);
-        this.channel.pipeline().get(MinecraftDecoder.class).setState(registry);
-    }
-}

src/main/java/io/minimum/minecraft/velocity/proxy/VelocityServer.java

@@ -1,6 +1,8 @@
 package io.minimum.minecraft.velocity.proxy;
 
+import io.minimum.minecraft.velocity.protocol.StateRegistry;
 import io.minimum.minecraft.velocity.protocol.netty.*;
+import io.minimum.minecraft.velocity.proxy.client.HandshakeSessionHandler;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.*;
@@ -32,9 +34,12 @@ public class VelocityServer {
                 .childHandler(new ChannelInitializer<Channel>() {
                     @Override
                     protected void initChannel(Channel ch) throws Exception {
-                        ch.attr(InboundMinecraftConnection.CONNECTION).set(new InboundMinecraftConnection(ch));
                         MinecraftPipelineUtils.strapPipelineForServer(ch);
-                        ch.pipeline().addLast("handler", new MinecraftClientSessionHandler());
+
+                        MinecraftConnection connection = new MinecraftConnection(ch);
+                        connection.setState(StateRegistry.HANDSHAKE);
+                        connection.setSessionHandler(new HandshakeSessionHandler(connection));
+                        ch.pipeline().addLast("handler", connection);
                     }
                 })
                 .bind(26671)
@@ -51,7 +56,7 @@ public class VelocityServer {
                 });
     }
 
-    Bootstrap initializeGenericBootstrap() {
+    public Bootstrap initializeGenericBootstrap() {
         return new Bootstrap()
                 .channel(NioSocketChannel.class)
                 .group(childGroup);

src/main/java/io/minimum/minecraft/velocity/proxy/backend/LoginSessionHandler.java

@@ -0,0 +1,41 @@
+package io.minimum.minecraft.velocity.proxy.backend;
+
+import io.minimum.minecraft.velocity.protocol.MinecraftPacket;
+import io.minimum.minecraft.velocity.protocol.StateRegistry;
+import io.minimum.minecraft.velocity.protocol.netty.MinecraftPipelineUtils;
+import io.minimum.minecraft.velocity.protocol.packets.Disconnect;
+import io.minimum.minecraft.velocity.protocol.packets.ServerLoginSuccess;
+import io.minimum.minecraft.velocity.protocol.packets.SetCompression;
+import io.minimum.minecraft.velocity.proxy.MinecraftSessionHandler;
+
+public class LoginSessionHandler implements MinecraftSessionHandler {
+    private final ServerConnection connection;
+
+    public LoginSessionHandler(ServerConnection connection) {
+        this.connection = connection;
+    }
+
+    @Override
+    public void handle(MinecraftPacket packet) {
+        if (packet instanceof Disconnect) {
+            Disconnect disconnect = (Disconnect) packet;
+            connection.disconnect();
+            connection.getProxyPlayer().handleConnectionException(connection.getServerInfo(), disconnect);
+        }
+
+        if (packet instanceof SetCompression) {
+            System.out.println("Enabling compression on server connection, this is inefficient!");
+            SetCompression sc = (SetCompression) packet;
+            connection.getChannel().setCompressionThreshold(sc.getThreshold());
+        }
+
+        if (packet instanceof ServerLoginSuccess) {
+            // the player has been logged on.
+            System.out.println("Player connected to remote server");
+            connection.getChannel().setState(StateRegistry.PLAY);
+            connection.getProxyPlayer().setConnectedServer(connection);
+            connection.getProxyPlayer().getConnection().setSessionHandler(new io.minimum.minecraft.velocity.proxy.client.PlaySessionHandler(connection.getProxyPlayer()));
+            connection.getChannel().setSessionHandler(new PlaySessionHandler(connection));
+        }
+    }
+}

src/main/java/io/minimum/minecraft/velocity/proxy/backend/PlaySessionHandler.java

@@ -0,0 +1,45 @@
+package io.minimum.minecraft.velocity.proxy.backend;
+
+import io.minimum.minecraft.velocity.protocol.MinecraftPacket;
+import io.minimum.minecraft.velocity.protocol.packets.Disconnect;
+import io.minimum.minecraft.velocity.protocol.packets.Ping;
+import io.minimum.minecraft.velocity.proxy.MinecraftSessionHandler;
+import io.netty.buffer.ByteBuf;
+import net.kyori.text.TextComponent;
+import net.kyori.text.format.TextColor;
+import net.kyori.text.serializer.ComponentSerializers;
+
+public class PlaySessionHandler implements MinecraftSessionHandler {
+    private final ServerConnection connection;
+
+    public PlaySessionHandler(ServerConnection connection) {
+        this.connection = connection;
+    }
+
+    @Override
+    public void handle(MinecraftPacket packet) {
+        if (packet instanceof Ping) {
+            // Make sure to reply back to the server so it doesn't think we're gone.
+            connection.getChannel().write(packet);
+            connection.getProxyPlayer().getConnection().write(packet);
+        } else if (packet instanceof Disconnect) {
+            // The server wants to disconnect us. TODO fallback handling
+            Disconnect original = (Disconnect) packet;
+            TextComponent reason = TextComponent.builder()
+                    .content("Disconnected from " + connection.getServerInfo().getName() + ":")
+                    .color(TextColor.RED)
+                    .append(TextComponent.of(" ", TextColor.WHITE))
+                    .append(ComponentSerializers.JSON.deserialize(original.getReason()))
+                    .build();
+            connection.getProxyPlayer().close(reason);
+        } else {
+            // Just forward the packet on. We don't have anything to handle at this time.
+            connection.getProxyPlayer().getConnection().write(packet);
+        }
+    }
+
+    @Override
+    public void handleUnknown(ByteBuf buf) {
+        connection.getProxyPlayer().getConnection().write(buf.retain());
+    }
+}

src/main/java/io/minimum/minecraft/velocity/proxy/backend/ServerConnection.java

@@ -0,0 +1,86 @@
+package io.minimum.minecraft.velocity.proxy.backend;
+
+import io.minimum.minecraft.velocity.protocol.StateRegistry;
+import io.minimum.minecraft.velocity.data.ServerInfo;
+import io.minimum.minecraft.velocity.protocol.netty.MinecraftPipelineUtils;
+import io.minimum.minecraft.velocity.protocol.packets.*;
+import io.minimum.minecraft.velocity.proxy.MinecraftConnection;
+import io.minimum.minecraft.velocity.proxy.VelocityServer;
+import io.minimum.minecraft.velocity.proxy.client.ConnectedPlayer;
+import io.netty.channel.*;
+
+public class ServerConnection {
+    private final ServerInfo serverInfo;
+    private final ConnectedPlayer proxyPlayer;
+    private final VelocityServer server;
+    private MinecraftConnection channel;
+
+    public ServerConnection(ServerInfo target, ConnectedPlayer proxyPlayer, VelocityServer server) {
+        this.serverInfo = target;
+        this.proxyPlayer = proxyPlayer;
+        this.server = server;
+    }
+
+    public void connect() {
+        server.initializeGenericBootstrap()
+                .handler(new ChannelInitializer<Channel>() {
+                    @Override
+                    protected void initChannel(Channel ch) throws Exception {
+                        MinecraftPipelineUtils.strapPipelineForProxy(ch);
+
+                        MinecraftConnection connection = new MinecraftConnection(ch);
+                        connection.setState(StateRegistry.HANDSHAKE);
+                        connection.setSessionHandler(new LoginSessionHandler(ServerConnection.this));
+                        ch.pipeline().addLast("handler", connection);
+                    }
+                })
+                .connect(serverInfo.getAddress())
+                .addListener(new ChannelFutureListener() {
+                    @Override
+                    public void operationComplete(ChannelFuture future) throws Exception {
+                        if (future.isSuccess()) {
+                            channel = future.channel().pipeline().get(MinecraftConnection.class);
+
+                            // Kick off the connection process
+                            startHandshake();
+                        } else {
+                            proxyPlayer.handleConnectionException(serverInfo, future.cause());
+                        }
+                    }
+                });
+    }
+
+    private void startHandshake() {
+        // Initiate a handshake.
+        Handshake handshake = new Handshake();
+        handshake.setNextStatus(2); // login
+        handshake.setProtocolVersion(proxyPlayer.getConnection().getProtocolVersion());
+        handshake.setServerAddress(serverInfo.getAddress().getHostString());
+        handshake.setPort(serverInfo.getAddress().getPort());
+        channel.write(handshake);
+
+        channel.setState(StateRegistry.LOGIN);
+
+        // Login
+        ServerLogin login = new ServerLogin();
+        login.setUsername(proxyPlayer.getUsername());
+        channel.write(login);
+    }
+
+    public ConnectedPlayer getProxyPlayer() {
+        return proxyPlayer;
+    }
+
+    public MinecraftConnection getChannel() {
+        return channel;
+    }
+
+    public ServerInfo getServerInfo() {
+        return serverInfo;
+    }
+
+    public void disconnect() {
+        channel.close();
+        channel = null;
+    }
+}

src/main/java/io/minimum/minecraft/velocity/proxy/client/ConnectedPlayer.java

@@ -1,8 +1,11 @@
-package io.minimum.minecraft.velocity.proxy;
+package io.minimum.minecraft.velocity.proxy.client;
 
 import io.minimum.minecraft.velocity.data.ServerInfo;
 import io.minimum.minecraft.velocity.protocol.packets.Chat;
 import io.minimum.minecraft.velocity.protocol.packets.Disconnect;
+import io.minimum.minecraft.velocity.proxy.MinecraftConnection;
+import io.minimum.minecraft.velocity.proxy.backend.ServerConnection;
+import io.minimum.minecraft.velocity.util.ThrowableUtils;
 import net.kyori.text.TextComponent;
 import net.kyori.text.format.TextColor;
 import net.kyori.text.serializer.ComponentSerializers;
@@ -12,10 +15,10 @@ import java.util.UUID;
 public class ConnectedPlayer {
     private final String username;
     private final UUID uniqueId;
-    private final InboundMinecraftConnection connection;
+    private final MinecraftConnection connection;
     private ServerConnection connectedServer;
 
-    public ConnectedPlayer(String username, UUID uniqueId, InboundMinecraftConnection connection) {
+    public ConnectedPlayer(String username, UUID uniqueId, MinecraftConnection connection) {
         this.username = username;
         this.uniqueId = uniqueId;
         this.connection = connection;
@@ -29,7 +32,7 @@ public class ConnectedPlayer {
         return uniqueId;
     }
 
-    public InboundMinecraftConnection getConnection() {
+    public MinecraftConnection getConnection() {
         return connection;
     }
 
@@ -38,8 +41,7 @@ public class ConnectedPlayer {
     }
 
     public void handleConnectionException(ServerInfo info, Throwable throwable) {
-        String error = String.format("%s: %s",
+        String error = ThrowableUtils.briefDescription(throwable);
-                throwable.getClass().getName(), throwable.getMessage());
         Disconnect disconnect = new Disconnect();
         disconnect.setReason(ComponentSerializers.JSON.serialize(TextComponent.of(error, TextColor.RED)));
         handleConnectionException(info, disconnect);
@@ -67,4 +69,10 @@ public class ConnectedPlayer {
     public void setConnectedServer(ServerConnection serverConnection) {
         this.connectedServer = serverConnection;
     }
+
+    public void close(TextComponent reason) {
+        Disconnect disconnect = new Disconnect();
+        disconnect.setReason(ComponentSerializers.JSON.serialize(reason));
+        connection.closeWith(disconnect);
+    }
 }

src/main/java/io/minimum/minecraft/velocity/proxy/client/HandshakeSessionHandler.java

@@ -0,0 +1,40 @@
+package io.minimum.minecraft.velocity.proxy.client;
+
+import com.google.common.base.Preconditions;
+import io.minimum.minecraft.velocity.protocol.MinecraftPacket;
+import io.minimum.minecraft.velocity.protocol.StateRegistry;
+import io.minimum.minecraft.velocity.protocol.packets.Handshake;
+import io.minimum.minecraft.velocity.proxy.MinecraftConnection;
+import io.minimum.minecraft.velocity.proxy.MinecraftSessionHandler;
+
+public class HandshakeSessionHandler implements MinecraftSessionHandler {
+    private final MinecraftConnection connection;
+
+    public HandshakeSessionHandler(MinecraftConnection connection) {
+        this.connection = Preconditions.checkNotNull(connection, "connection");
+    }
+
+    @Override
+    public void handle(MinecraftPacket packet) {
+        if (!(packet instanceof Handshake)) {
+            throw new IllegalArgumentException("Did not expect packet " + packet.getClass().getName());
+        }
+
+        Handshake handshake = (Handshake) packet;
+        connection.setProtocolVersion(handshake.getProtocolVersion());
+        switch (handshake.getNextStatus()) {
+            case 1:
+                // Status protocol
+                connection.setState(StateRegistry.STATUS);
+                connection.setSessionHandler(new StatusSessionHandler(connection));
+                break;
+            case 2:
+                connection.setState(StateRegistry.LOGIN);
+                connection.setSessionHandler(new LoginSessionHandler(connection));
+                break;
+            default:
+                throw new IllegalArgumentException("Invalid state " + handshake.getNextStatus());
+        }
+
+    }
+}

src/main/java/io/minimum/minecraft/velocity/proxy/client/InitialConnectSessionHandler.java

@@ -0,0 +1,23 @@
+package io.minimum.minecraft.velocity.proxy.client;
+
+import io.minimum.minecraft.velocity.protocol.MinecraftPacket;
+import io.minimum.minecraft.velocity.proxy.MinecraftSessionHandler;
+
+public class InitialConnectSessionHandler implements MinecraftSessionHandler {
+    private final ConnectedPlayer player;
+
+    public InitialConnectSessionHandler(ConnectedPlayer player) {
+        this.player = player;
+    }
+
+    @Override
+    public void handle(MinecraftPacket packet) {
+        // No-op: will never handle packets
+    }
+
+    @Override
+    public void disconnected() {
+        // the user cancelled the login process
+        player.getConnectedServer().disconnect();
+    }
+}

src/main/java/io/minimum/minecraft/velocity/proxy/client/LoginSessionHandler.java

@@ -1,22 +1,23 @@
-package io.minimum.minecraft.velocity.proxy.handler;
+package io.minimum.minecraft.velocity.proxy.client;
 
 import com.google.common.base.Preconditions;
 import io.minimum.minecraft.velocity.data.ServerInfo;
 import io.minimum.minecraft.velocity.protocol.MinecraftPacket;
+import io.minimum.minecraft.velocity.protocol.StateRegistry;
 import io.minimum.minecraft.velocity.protocol.packets.ServerLogin;
 import io.minimum.minecraft.velocity.protocol.packets.ServerLoginSuccess;
-import io.minimum.minecraft.velocity.protocol.packets.SetCompression;
 import io.minimum.minecraft.velocity.proxy.*;
+import io.minimum.minecraft.velocity.proxy.backend.ServerConnection;
 
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
 import java.util.UUID;
 
 public class LoginSessionHandler implements MinecraftSessionHandler {
-    private final InboundMinecraftConnection connection;
+    private final MinecraftConnection inbound;
 
-    public LoginSessionHandler(InboundMinecraftConnection connection) {
+    public LoginSessionHandler(MinecraftConnection inbound) {
-        this.connection = Preconditions.checkNotNull(connection, "connection");
+        this.inbound = Preconditions.checkNotNull(inbound, "inbound");
     }
 
     @Override
@@ -24,15 +25,22 @@ public class LoginSessionHandler implements MinecraftSessionHandler {
         Preconditions.checkArgument(packet instanceof ServerLogin, "Expected a ServerLogin packet, not " + packet.getClass().getName());
 
         // TODO: Encryption
-        connection.enableCompression();
+        inbound.setCompressionThreshold(256);
 
         String username = ((ServerLogin) packet).getUsername();
         ServerLoginSuccess success = new ServerLoginSuccess();
         success.setUsername(username);
         success.setUuid(generateOfflinePlayerUuid(username));
-        connection.write(success);
+        inbound.write(success);
 
-        connection.initiatePlay(success);
+        // Initiate a regular connection and move over to it.
+        ConnectedPlayer player = new ConnectedPlayer(success.getUsername(), success.getUuid(), inbound);
+        ServerInfo info = new ServerInfo("test", new InetSocketAddress("localhost", 25565));
+        ServerConnection connection = new ServerConnection(info, player, VelocityServer.getServer());
+
+        inbound.setState(StateRegistry.PLAY);
+        inbound.setSessionHandler(new InitialConnectSessionHandler(player));
+        connection.connect();
     }
 
     private static UUID generateOfflinePlayerUuid(String username) {

src/main/java/io/minimum/minecraft/velocity/proxy/client/PlaySessionHandler.java

@@ -1,20 +1,16 @@
-package io.minimum.minecraft.velocity.proxy.handler;
+package io.minimum.minecraft.velocity.proxy.client;
 
 import io.minimum.minecraft.velocity.protocol.MinecraftPacket;
 import io.minimum.minecraft.velocity.protocol.packets.Chat;
 import io.minimum.minecraft.velocity.protocol.packets.Ping;
-import io.minimum.minecraft.velocity.proxy.ConnectedPlayer;
 import io.minimum.minecraft.velocity.proxy.MinecraftSessionHandler;
-import io.minimum.minecraft.velocity.proxy.ServerConnection;
 import io.netty.buffer.ByteBuf;
 
 public class PlaySessionHandler implements MinecraftSessionHandler {
     private final ConnectedPlayer player;
-    private final ServerConnection connection;
 
-    public PlaySessionHandler(ConnectedPlayer player, ServerConnection connection) {
+    public PlaySessionHandler(ConnectedPlayer player) {
         this.player = player;
-        this.connection = connection;
     }
 
     @Override
@@ -25,14 +21,17 @@ public class PlaySessionHandler implements MinecraftSessionHandler {
             return;
         }
 
-        if (packet instanceof Chat) {
+        // If we don't want to handle this packet, just forward it on.
-            // TODO: handle this ourselves, for now do this
+        player.getConnectedServer().getChannel().write(packet);
-            player.getConnectedServer().forward(packet);
-        }
     }
 
     @Override
     public void handleUnknown(ByteBuf buf) {
-        connection.forward(buf.retain());
+        player.getConnectedServer().getChannel().write(buf.retain());
+    }
+
+    @Override
+    public void disconnected() {
+        player.getConnectedServer().disconnect();
     }
 }

src/main/java/io/minimum/minecraft/velocity/proxy/client/StatusSessionHandler.java

@@ -1,15 +1,14 @@
-package io.minimum.minecraft.velocity.proxy.handler;
+package io.minimum.minecraft.velocity.proxy.client;
 
 import com.google.common.base.Preconditions;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import io.minimum.minecraft.velocity.data.ServerPing;
 import io.minimum.minecraft.velocity.protocol.MinecraftPacket;
-import io.minimum.minecraft.velocity.protocol.packets.LegacyPing;
 import io.minimum.minecraft.velocity.protocol.packets.Ping;
 import io.minimum.minecraft.velocity.protocol.packets.StatusRequest;
 import io.minimum.minecraft.velocity.protocol.packets.StatusResponse;
-import io.minimum.minecraft.velocity.proxy.InboundMinecraftConnection;
+import io.minimum.minecraft.velocity.proxy.MinecraftConnection;
 import io.minimum.minecraft.velocity.proxy.MinecraftSessionHandler;
 import net.kyori.text.Component;
 import net.kyori.text.TextComponent;
@@ -19,9 +18,9 @@ public class StatusSessionHandler implements MinecraftSessionHandler {
     private static final Gson GSON = new GsonBuilder()
             .registerTypeHierarchyAdapter(Component.class, new GsonComponentSerializer())
             .create();
-    private final InboundMinecraftConnection connection;
+    private final MinecraftConnection connection;
 
-    public StatusSessionHandler(InboundMinecraftConnection connection) {
+    public StatusSessionHandler(MinecraftConnection connection) {
         this.connection = connection;
     }
 

src/main/java/io/minimum/minecraft/velocity/proxy/handler/HandshakeSessionHandler.java

@@ -1,25 +0,0 @@
-package io.minimum.minecraft.velocity.proxy.handler;
-
-import com.google.common.base.Preconditions;
-import io.minimum.minecraft.velocity.protocol.MinecraftPacket;
-import io.minimum.minecraft.velocity.protocol.packets.Handshake;
-import io.minimum.minecraft.velocity.proxy.InboundMinecraftConnection;
-import io.minimum.minecraft.velocity.proxy.MinecraftSessionHandler;
-
-public class HandshakeSessionHandler implements MinecraftSessionHandler {
-    private final InboundMinecraftConnection connection;
-
-    public HandshakeSessionHandler(InboundMinecraftConnection connection) {
-        this.connection = Preconditions.checkNotNull(connection, "connection");
-    }
-
-    @Override
-    public void handle(MinecraftPacket packet) {
-        if (!(packet instanceof Handshake)) {
-            throw new IllegalArgumentException("Did not expect packet " + packet.getClass().getName());
-        }
-
-        Handshake handshake = (Handshake) packet;
-        connection.handleHandshake(handshake);
-    }
-}

src/main/java/io/minimum/minecraft/velocity/util/ThrowableUtils.java

@@ -0,0 +1,9 @@
+package io.minimum.minecraft.velocity.util;
+
+public enum ThrowableUtils {
+    ;
+
+    public static String briefDescription(Throwable throwable) {
+        return throwable.getClass().getName() + ": " + throwable.getMessage();
+    }
+}