Initial Netty implementation for networking

Split into 1 LoginServer and 1 ChannelServer per channel.

There is still a lot of cleanup and refactoring to be done.
Currently, the reliance on IoSession holding client state
is the most pressing issue to be addressed.

src/main/java/net/PacketProcessor.java

@@ -21,6 +21,7 @@
  */
 package net;
 
+import net.netty.LoginServer;
 import net.opcodes.RecvOpcode;
 import net.server.channel.handlers.*;
 import net.server.handlers.CustomPacketHandler;
@@ -47,7 +48,7 @@ public final class PacketProcessor {
     }
 
     public static PacketProcessor getLoginServerProcessor() {
-        return getProcessor(-1, -1);
+        return getProcessor(LoginServer.WORLD, LoginServer.CHANNEL);
     }
 
     public static PacketProcessor getChannelServerProcessor(int world, int channel) {

src/main/java/net/netty/AbstractServer.java

@@ -0,0 +1,12 @@
+package net.netty;
+
+public abstract class AbstractServer {
+    final int port;
+
+    AbstractServer(int port) {
+        this.port = port;
+    }
+
+    public abstract void start();
+    public abstract void stop();
+}

src/main/java/net/netty/ChannelServer.java

@@ -0,0 +1,40 @@
+package net.netty;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+
+public class ChannelServer extends AbstractServer {
+    private final int world;
+    private final int channel;
+    private Channel nettyChannel;
+
+    public ChannelServer(int port, int world, int channel) {
+        super(port);
+        this.world = world;
+        this.channel = channel;
+    }
+
+    @Override
+    public void start() {
+        EventLoopGroup parentGroup = new NioEventLoopGroup();
+        EventLoopGroup childGroup = new NioEventLoopGroup();
+        ServerBootstrap bootstrap = new ServerBootstrap()
+                .group(parentGroup, childGroup)
+                .channel(NioServerSocketChannel.class)
+                .childHandler(new ChannelServerInitializer(world, channel));
+
+        this.nettyChannel = bootstrap.bind(port).syncUninterruptibly().channel();
+    }
+
+    @Override
+    public void stop() {
+        if (nettyChannel == null) {
+            throw new IllegalStateException("Must start ChannelServer before stopping it");
+        }
+
+        nettyChannel.close().syncUninterruptibly();
+    }
+}

src/main/java/net/netty/ChannelServerInitializer.java

@@ -0,0 +1,31 @@
+package net.netty;
+
+import client.MapleClient;
+import io.netty.channel.socket.SocketChannel;
+import net.PacketProcessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ChannelServerInitializer extends ServerChannelInitializer {
+    private static final Logger log = LoggerFactory.getLogger(ChannelServerInitializer.class);
+
+    private final int world;
+    private final int channel;
+
+    public ChannelServerInitializer(int world, int channel) {
+        this.world = world;
+        this.channel = channel;
+    }
+
+    @Override
+    public void initChannel(SocketChannel socketChannel) {
+        final String clientIp = socketChannel.remoteAddress().getHostName();
+        log.debug("Client connected to world {}, channel {} from {}", world, channel, clientIp);
+
+        PacketProcessor packetProcessor = PacketProcessor.getChannelServerProcessor(world, channel);
+        final MapleClient client = new MapleClient(packetProcessor, world, channel);
+        client.setSessionId(sessionId.getAndIncrement());
+
+        initPipeline(socketChannel, client);
+    }
+}

src/main/java/net/netty/ClientCyphers.java

@@ -12,9 +12,9 @@ public class ClientCyphers {
         this.receive = receive;
     }
 
-    public static ClientCyphers generateNew() {
-        MapleAESOFB send = new MapleAESOFB(InitializationVector.generateSend(), ServerConstants.VERSION);
-        MapleAESOFB receive = new MapleAESOFB(InitializationVector.generateReceive(), ServerConstants.VERSION);
+    public static ClientCyphers of(InitializationVector sendIv, InitializationVector receiveIv) {
+        MapleAESOFB send = new MapleAESOFB(sendIv, (short) (0xFFFF - ServerConstants.VERSION));
+        MapleAESOFB receive = new MapleAESOFB(receiveIv, ServerConstants.VERSION);
         return new ClientCyphers(send, receive);
     }
 

src/main/java/net/netty/ClientInitializer.java

@@ -1,20 +0,0 @@
-package net.netty;
-
-import client.MapleClient;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.socket.SocketChannel;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class ClientInitializer extends ChannelInitializer<SocketChannel> {
-    private static final Logger log = LoggerFactory.getLogger(ClientInitializer.class);
-
-    @Override
-    public void initChannel(SocketChannel socketChannel) {
-        final String clientIp = socketChannel.remoteAddress().getHostName();
-        log.debug("Client initiated new connection from: {}", clientIp);
-
-        socketChannel.pipeline().addLast("PacketCodec", new PacketCodec(ClientCyphers.generateNew()));
-        socketChannel.pipeline().addLast("MapleClient", new MapleClient());
-    }
-}

src/main/java/net/netty/LoginServer.java

@@ -0,0 +1,38 @@
+package net.netty;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+
+public class LoginServer extends AbstractServer {
+    public static final int WORLD = -1;
+    public static final int CHANNEL = -1;
+    private Channel channel;
+
+    public LoginServer(int port) {
+        super(port);
+    }
+
+    @Override
+    public void start() {
+        EventLoopGroup parentGroup = new NioEventLoopGroup();
+        EventLoopGroup childGroup = new NioEventLoopGroup();
+        ServerBootstrap bootstrap = new ServerBootstrap()
+                .group(parentGroup, childGroup)
+                .channel(NioServerSocketChannel.class)
+                .childHandler(new LoginServerInitializer());
+
+        this.channel = bootstrap.bind(port).syncUninterruptibly().channel();
+    }
+
+    @Override
+    public void stop() {
+        if (channel == null) {
+            throw new IllegalStateException("Must start LoginServer before stopping it");
+        }
+
+        channel.close().syncUninterruptibly();
+    }
+}

src/main/java/net/netty/LoginServerInitializer.java

@@ -0,0 +1,23 @@
+package net.netty;
+
+import client.MapleClient;
+import io.netty.channel.socket.SocketChannel;
+import net.PacketProcessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LoginServerInitializer extends ServerChannelInitializer {
+    private static final Logger log = LoggerFactory.getLogger(LoginServerInitializer.class);
+
+    @Override
+    public void initChannel(SocketChannel socketChannel) {
+        final String clientIp = socketChannel.remoteAddress().getHostName();
+        log.debug("Client connected to login server from {} ", clientIp);
+
+        PacketProcessor packetProcessor = PacketProcessor.getLoginServerProcessor();
+        final MapleClient client = new MapleClient(packetProcessor, LoginServer.WORLD, LoginServer.CHANNEL);
+        client.setSessionId(sessionId.getAndIncrement());
+
+        initPipeline(socketChannel, client);
+    }
+}

src/main/java/net/netty/PacketDecoder.java

@@ -1,9 +1,11 @@
 package net.netty;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ReplayingDecoder;
 import net.mina.MapleCustomEncryption;
+import net.packet.ByteBufInPacket;
 import tools.MapleAESOFB;
 
 import java.util.List;
@@ -28,7 +30,7 @@ public class PacketDecoder extends ReplayingDecoder<Void> {
         in.readBytes(packet);
         receiveCypher.crypt(packet);
         MapleCustomEncryption.decryptData(packet);
-        out.add(packet);
+        out.add(new ByteBufInPacket(Unpooled.wrappedBuffer(packet)));
         // TODO conditionally log the packet
     }
 

src/main/java/net/netty/ServerChannelInitializer.java

@@ -0,0 +1,22 @@
+package net.netty;
+
+import client.MapleClient;
+import constants.net.ServerConstants;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.SocketChannel;
+import tools.MaplePacketCreator;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+public abstract class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
+    static final AtomicLong sessionId = new AtomicLong(7777);
+
+    void initPipeline(SocketChannel socketChannel, MapleClient client) {
+        final InitializationVector sendIv = InitializationVector.generateSend();
+        final InitializationVector recvIv = InitializationVector.generateReceive();
+        socketChannel.writeAndFlush(Unpooled.wrappedBuffer(MaplePacketCreator.getHello(ServerConstants.VERSION, sendIv, recvIv)));
+        socketChannel.pipeline().addLast("PacketCodec", new PacketCodec(ClientCyphers.of(sendIv, recvIv)));
+        socketChannel.pipeline().addLast("MapleClient", client);
+    }
+}

src/main/java/net/server/Server.java

@@ -37,6 +37,7 @@ import constants.net.OpcodeConstants;
 import constants.net.ServerConstants;
 import net.MapleServerHandler;
 import net.mina.MapleCodecFactory;
+import net.netty.LoginServer;
 import net.server.audit.ThreadTracker;
 import net.server.audit.locks.MonitoredLockType;
 import net.server.audit.locks.MonitoredReadLock;
@@ -106,6 +107,7 @@ public class Server {
     private static final List<Integer> activeCoupons = new LinkedList<>();
 
     private IoAcceptor acceptor;
+    private LoginServer loginServer;
     private List<Map<Integer, String>> channels = new LinkedList<>();
     private List<World> worlds = new ArrayList<>();
     private final Properties subnetInfo = new Properties();
@@ -906,17 +908,8 @@ public class Server {
             }
         }
 
-        IoBuffer.setUseDirectBuffer(false);     // join IO operations performed by lxconan
-        IoBuffer.setAllocator(new SimpleBufferAllocator());
-        acceptor = new NioSocketAcceptor();
-        acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new MapleCodecFactory()));
-        acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 30);
-        acceptor.setHandler(new MapleServerHandler());
-        try {
-            acceptor.bind(new InetSocketAddress(8484));
-        } catch (IOException ex) {
-            ex.printStackTrace();
-        }
+        // acceptor = initAcceptor(8484);
+        loginServer = initLoginServer(8484);
 
         log.info("Listening on port 8484");
 
@@ -932,6 +925,28 @@ public class Server {
         }
     }
 
+    private LoginServer initLoginServer(int port) {
+        LoginServer loginServer = new LoginServer(port);
+        loginServer.start();
+        return loginServer;
+    }
+
+    private IoAcceptor initAcceptor(int port) {
+        IoBuffer.setUseDirectBuffer(false);     // join IO operations performed by lxconan
+        IoBuffer.setAllocator(new SimpleBufferAllocator());
+        acceptor = new NioSocketAcceptor();
+        acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new MapleCodecFactory()));
+        acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 30);
+        acceptor.setHandler(new MapleServerHandler());
+        try {
+            acceptor.bind(new InetSocketAddress(port));
+        } catch (IOException ex) {
+            ex.printStackTrace();
+        }
+
+        return acceptor;
+    }
+
     private static void setAllLoggedOut(Connection con) throws SQLException {
         try (PreparedStatement ps = con.prepareStatement("UPDATE accounts SET loggedin = 0")) {
             ps.executeUpdate();

src/main/java/net/server/channel/Channel.java

@@ -26,6 +26,7 @@ import config.YamlConfig;
 import constants.game.GameConstants;
 import net.MapleServerHandler;
 import net.mina.MapleCodecFactory;
+import net.netty.ChannelServer;
 import net.server.PlayerStorage;
 import net.server.Server;
 import net.server.audit.LockCollector;
@@ -58,6 +59,7 @@ import tools.MaplePacketCreator;
 import tools.Pair;
 
 import java.io.File;
+import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.*;
 import java.util.Map.Entry;
@@ -69,6 +71,7 @@ public final class Channel {
     private int port = 7575;
     private PlayerStorage players = new PlayerStorage();
     private int world, channel;
+    private ChannelServer channelServer;
     private IoAcceptor acceptor;
     private String ip, serverMessage;
     private MapleMapManager mapManager;
@@ -122,14 +125,8 @@ public final class Channel {
             port = 7575 + this.channel - 1;
             port += (world * 100);
             ip = YamlConfig.config.server.HOST + ":" + port;
-            IoBuffer.setUseDirectBuffer(false);
-            IoBuffer.setAllocator(new SimpleBufferAllocator());
-            acceptor = new NioSocketAcceptor();
-            acceptor.setHandler(new MapleServerHandler(world, channel));
-            acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 30);
-            acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new MapleCodecFactory()));
-            acceptor.bind(new InetSocketAddress(port));
-            ((SocketSessionConfig) acceptor.getSessionConfig()).setTcpNoDelay(true);
+            // acceptor = initAcceptor();
+            channelServer = initServer(port, world, channel);
             expedType.addAll(Arrays.asList(MapleExpeditionType.values()));
             
             if (Server.getInstance().isOnline()) {  // postpone event loading to improve boot time... thanks Riizade, daronhudson for noticing slow startup times
@@ -156,6 +153,24 @@ public final class Channel {
             e.printStackTrace();
         }
     }
+
+    private IoAcceptor initAcceptor() throws IOException {
+        IoBuffer.setUseDirectBuffer(false);
+        IoBuffer.setAllocator(new SimpleBufferAllocator());
+        IoAcceptor acceptor = new NioSocketAcceptor();
+        acceptor.setHandler(new MapleServerHandler(world, channel));
+        acceptor.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, 30);
+        acceptor.getFilterChain().addLast("codec", new ProtocolCodecFilter(new MapleCodecFactory()));
+        acceptor.bind(new InetSocketAddress(port));
+        ((SocketSessionConfig) acceptor.getSessionConfig()).setTcpNoDelay(true);
+        return acceptor;
+    }
+
+    private ChannelServer initServer(int port, int world, int channel) {
+        this.channelServer = new ChannelServer(port, world, channel);
+        channelServer.start();
+        return channelServer;
+    }
     
     public synchronized void reloadEventScriptManager(){
         if (finishedShutdown) {