Fixed startup issue with StaticPvpZoneLoader and changed the network code to use built-in Java classes for async i/o

bitbucket-pipelines.yml

@@ -1,4 +1,3 @@
-image: openjdk:12
 pipelines:
   default:
     - step:
@@ -11,7 +10,7 @@ pipelines:
           - client-holocore/**
     - step:
         name: test, build, and deploy
-        #image: openjdk:12
+        image: openjdk:12
         caches:
           - docker
           - gradle

src/main/java/com/projectswg/holocore/resources/support/data/server_info/loader/StaticPvpZoneLoader.java

@@ -65,7 +65,7 @@ public class StaticPvpZoneLoader extends DataLoader {
 		private final double radius;
 		
 		public StaticPvpZoneInfo(SdbLoader.SdbResultSet set) {
-			id = (int) set.getInt("id");
+			id = (int) set.getInt("pvp_zone_id");
 			
 			location = Location.builder()
 					.setX(set.getInt("x"))

src/main/java/com/projectswg/holocore/resources/support/global/network/NetworkClient.kt

@@ -47,14 +47,16 @@ import me.joshlarson.jlcommon.log.Log
 import java.io.IOException
 import java.net.InetSocketAddress
 import java.net.SocketAddress
-import java.nio.ByteBuffer
-import java.nio.channels.SocketChannel
+import java.nio.channels.AsynchronousCloseException
+import java.nio.channels.AsynchronousSocketChannel
+import java.nio.channels.ClosedChannelException
+import java.nio.channels.CompletionHandler
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.atomic.AtomicReference
 import java.util.function.Consumer
 
-class NetworkClient(private val socket: SocketChannel) {
+class NetworkClient(private val socket: AsynchronousSocketChannel) {
 	
 	private val remoteAddress: SocketAddress? = try { socket.remoteAddress } catch (e: IOException) { null }
 	private val inboundBuffer: NetBuffer
@@ -99,33 +101,39 @@ class NetworkClient(private val socket: SocketChannel) {
 		return "NetworkClient[$remoteAddress]"
 	}
 	
-	fun addToInbound(data: ByteBuffer) {
-		synchronized(inboundBuffer) {
-			if (data.remaining() > inboundBuffer.remaining()) {
-				StandardLog.onPlayerError(this, player, "Possible hack attempt detected with buffer overflow.  Closing connection to $remoteAddress")
-				close(ConnectionStoppedReason.APPLICATION)
-				return
-			}
-			try {
-				inboundBuffer.add(data)
-				inboundBuffer.flip()
-				while (NetworkProtocol.canDecode(inboundBuffer)) {
-					val p = NetworkProtocol.decode(inboundBuffer)
-					if (p == null || !allowInbound(p))
-						continue
-					p.socketAddress = remoteAddress
-					processPacket(p)
-					intentChain.broadcastAfter(InboundPacketIntent(player, p))
+	private fun startRead() {
+		socket.read(inboundBuffer.buffer, null, object : CompletionHandler<Int, Any?> {
+			override fun completed(result: Int?, attachment: Any?) {
+				try {
+					inboundBuffer.flip()
+					while (NetworkProtocol.canDecode(inboundBuffer)) {
+						val p = NetworkProtocol.decode(inboundBuffer)
+						if (p == null || !allowInbound(p))
+							continue
+						p.socketAddress = remoteAddress
+						processPacket(p)
+						intentChain.broadcastAfter(InboundPacketIntent(player, p))
+					}
+					inboundBuffer.compact()
+				} catch (e: HolocoreSessionException) {
+					onSessionError(e)
+				} catch (e: IOException) {
+					Log.w("Failed to process inbound packets. IOException: %s", e.message)
+					close()
 				}
-				inboundBuffer.compact()
-			} catch (e: HolocoreSessionException) {
-				onSessionError(e)
-			} catch (e: IOException) {
-				Log.w("Failed to process inbound packets. IOException: %s", e.message)
-				close()
+				
+				startRead()
 			}
 			
-		}
+			override fun failed(exc: Throwable?, attachment: Any?) {
+				if (exc != null) {
+					if (exc !is AsynchronousCloseException && exc !is ClosedChannelException)
+						Log.w(exc)
+					close()
+				}
+			}
+			
+		})
 	}
 	
 	private fun addToOutbound(p: SWGPacket) {
@@ -134,12 +142,12 @@ class NetworkClient(private val socket: SocketChannel) {
 				try {
 					val buffer = NetworkProtocol.encode(p).buffer
 					while (connected.get() && buffer.hasRemaining()) {
-						socket.write(buffer)
+						socket.write(buffer).get()
 						if (buffer.hasRemaining())
 							Delay.sleepMilli(1)
 					}
-				} catch (e: IOException) {
-					StandardLog.onPlayerError(this, player, "failed to write network data. ${e.javaClass.name}: ${e.message}")
+				} catch (t: Throwable) {
+					StandardLog.onPlayerError(this, player, "failed to write network data. ${t.javaClass.name}: ${t.message}")
 					close()
 				}
 			}
@@ -150,6 +158,7 @@ class NetworkClient(private val socket: SocketChannel) {
 		StandardLog.onPlayerTrace(this, player, "connecting")
 		status.set(SessionStatus.CONNECTING)
 		intentChain.broadcastAfter(ConnectionOpenedIntent(player))
+		startRead()
 	}
 	
 	private fun onConnected() {

src/main/java/com/projectswg/holocore/services/support/global/network/NetworkClientService.java

@@ -1,216 +0,0 @@
-/***********************************************************************************
- * Copyright (c) 2018 /// Project SWG /// www.projectswg.com                       *
- *                                                                                 *
- * ProjectSWG is the first NGE emulator for Star Wars Galaxies founded on          *
- * July 7th, 2011 after SOE announced the official shutdown of Star Wars Galaxies. *
- * Our goal is to create an emulator which will provide a server for players to    *
- * continue playing a game similar to the one they used to play. We are basing     *
- * it on the final publish of the game prior to end-game events.                   *
- *                                                                                 *
- * This file is part of Holocore.                                                  *
- *                                                                                 *
- * --------------------------------------------------------------------------------*
- *                                                                                 *
- * Holocore is free software: you can redistribute it and/or modify                *
- * it under the terms of the GNU Affero General Public License as                  *
- * published by the Free Software Foundation, either version 3 of the              *
- * License, or (at your option) any later version.                                 *
- *                                                                                 *
- * Holocore is distributed in the hope that it will be useful,                     *
- * but WITHOUT ANY WARRANTY; without even the implied warranty of                  *
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the                   *
- * GNU Affero General Public License for more details.                             *
- *                                                                                 *
- * You should have received a copy of the GNU Affero General Public License        *
- * along with Holocore.  If not, see <http://www.gnu.org/licenses/>.               *
- ***********************************************************************************/
-package com.projectswg.holocore.services.support.global.network;
-
-import com.projectswg.common.network.NetBuffer;
-import com.projectswg.common.network.packets.swg.holo.HoloConnectionStopped.ConnectionStoppedReason;
-import com.projectswg.holocore.ProjectSWG;
-import com.projectswg.holocore.ProjectSWG.CoreException;
-import com.projectswg.holocore.intents.support.global.network.CloseConnectionIntent;
-import com.projectswg.holocore.intents.support.global.network.ConnectionClosedIntent;
-import com.projectswg.holocore.resources.support.data.server_info.StandardLog;
-import com.projectswg.holocore.resources.support.data.server_info.mongodb.PswgDatabase;
-import com.projectswg.holocore.resources.support.global.network.NetworkClient;
-import com.projectswg.holocore.resources.support.global.network.UDPServer;
-import com.projectswg.holocore.resources.support.global.network.UDPServer.UDPPacket;
-import com.projectswg.holocore.resources.support.global.player.Player;
-import me.joshlarson.jlcommon.concurrency.BasicThread;
-import me.joshlarson.jlcommon.control.IntentHandler;
-import me.joshlarson.jlcommon.control.Service;
-import me.joshlarson.jlcommon.log.Log;
-
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
-import java.nio.channels.ServerSocketChannel;
-import java.nio.channels.SocketChannel;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-public class NetworkClientService extends Service {
-	
-	private static final int INBOUND_BUFFER_SIZE = 4096;
-	
-	private final ServerSocketChannel tcpServer;
-	private final BasicThread acceptThreadPool;
-	private final Map<Long, NetworkClient> clients;
-	private final ByteBuffer inboundBuffer;
-	private final UDPServer udpServer;
-	private volatile boolean operational;
-	
-	public NetworkClientService() {
-		this.acceptThreadPool = new BasicThread("network-client-accept", this::acceptLoop);
-		this.clients = new ConcurrentHashMap<>();
-		this.inboundBuffer = ByteBuffer.allocate(INBOUND_BUFFER_SIZE);
-		this.operational = true;
-		{
-			int bindPort = getBindPort();
-			try {
-				tcpServer = ServerSocketChannel.open();
-				tcpServer.bind(new InetSocketAddress(bindPort), 64);
-				tcpServer.configureBlocking(false);
-				udpServer = new UDPServer(bindPort, 32);
-			} catch (IOException e) {
-				throw new CoreException("Failed to start networking", e);
-			}
-			udpServer.setCallback(this::onUdpPacket);
-		}
-	}
-	
-	@Override
-	public boolean start() {
-		acceptThreadPool.start();
-		return true;
-	}
-	
-	@Override
-	public boolean isOperational() {
-		return operational;
-	}
-	
-	@Override
-	public boolean stop() {
-		for (NetworkClient client : clients.values())
-			client.close(ConnectionStoppedReason.APPLICATION);
-		
-		try {
-			tcpServer.close();
-		} catch (IOException e) {
-			Log.w("Failed to close TCP server");
-		}
-		acceptThreadPool.stop(true);
-		return acceptThreadPool.awaitTermination(1000);
-	}
-	
-	@Override
-	public boolean terminate() {
-		udpServer.close();
-		return super.terminate();
-	}
-	
-	private void acceptLoop() {
-		try (Selector selector = Selector.open()) {
-			tcpServer.register(selector, SelectionKey.OP_ACCEPT);
-			while (tcpServer.isOpen()) {
-				selector.select();
-				Iterator<SelectionKey> it = selector.selectedKeys().iterator();
-				while (it.hasNext()) {
-					SelectionKey key = it.next();
-					if (key.isAcceptable()) {
-						acceptConnection(selector);
-					}
-					if (key.isReadable()) {
-						read(key);
-					}
-					it.remove();
-				}
-			}
-		} catch (IOException e) {
-			Log.a(e);
-		} finally {
-			operational = false;
-		}
-	}
-	
-	private void acceptConnection(Selector selector) {
-		try {
-			SocketChannel client = tcpServer.accept();
-			if (client != null) {
-				client.configureBlocking(false);
-				NetworkClient networkClient = new NetworkClient(client);
-				client.register(selector, SelectionKey.OP_READ, networkClient);
-				clients.put(networkClient.getId(), networkClient);
-			}
-		} catch (Throwable t) {
-			Log.w("%s: Failed to accept connection", getClass().getSimpleName());
-		}
-	}
-	
-	private void read(SelectionKey key) {
-		Player player = null;
-		try {
-			SocketChannel channel = (SocketChannel) key.channel();
-			NetworkClient client = (NetworkClient) key.attachment();
-			player = client.getPlayer();
-			inboundBuffer.clear();
-			channel.read(inboundBuffer);
-			inboundBuffer.flip();
-			client.addToInbound(inboundBuffer);
-		} catch (Throwable t) {
-			if (player != null)
-				StandardLog.onPlayerError(this, player, "failed to read data");
-			else
-				Log.w("%s: Failed to read data", getClass().getSimpleName());
-		}
-	}
-	
-	private void disconnect(long networkId) {
-		disconnect(clients.get(networkId));
-	}
-	
-	private void disconnect(NetworkClient client) {
-		if (client == null)
-			return;
-		
-		client.close(ConnectionStoppedReason.APPLICATION);
-	}
-	
-	private void onUdpPacket(UDPPacket packet) {
-		if (packet.getLength() <= 0)
-			return;
-		if (packet.getData()[0] == 1) {
-			sendState(packet.getAddress(), packet.getPort());
-		}
-	}
-	
-	private void sendState(InetAddress addr, int port) {
-		String status = ProjectSWG.getGalaxy().getStatus().name();
-		NetBuffer data = NetBuffer.allocate(3 + status.length());
-		data.addByte(1);
-		data.addAscii(status);
-		udpServer.send(port, addr, data.array());
-	}
-	
-	@IntentHandler
-	private void handleCloseConnectionIntent(CloseConnectionIntent ccii) {
-		disconnect(ccii.getPlayer().getNetworkId());
-	}
-	
-	@IntentHandler
-	private void handleConnectionClosedIntent(ConnectionClosedIntent cci) {
-		disconnect(cci.getPlayer().getNetworkId());
-	}
-	
-	private int getBindPort() {
-		return PswgDatabase.INSTANCE.getConfig().getInt(this, "bindPort", 44463);
-	}
-	
-}

src/main/java/com/projectswg/holocore/services/support/global/network/NetworkClientService.kt

@@ -0,0 +1,165 @@
+/***********************************************************************************
+ * Copyright (c) 2018 /// Project SWG /// www.projectswg.com                       *
+ * *
+ * ProjectSWG is the first NGE emulator for Star Wars Galaxies founded on          *
+ * July 7th, 2011 after SOE announced the official shutdown of Star Wars Galaxies. *
+ * Our goal is to create an emulator which will provide a server for players to    *
+ * continue playing a game similar to the one they used to play. We are basing     *
+ * it on the final publish of the game prior to end-game events.                   *
+ * *
+ * This file is part of Holocore.                                                  *
+ * *
+ * --------------------------------------------------------------------------------*
+ * *
+ * Holocore is free software: you can redistribute it and/or modify                *
+ * it under the terms of the GNU Affero General Public License as                  *
+ * published by the Free Software Foundation, either version 3 of the              *
+ * License, or (at your option) any later version.                                 *
+ * *
+ * Holocore is distributed in the hope that it will be useful,                     *
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of                  *
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the                   *
+ * GNU Affero General Public License for more details.                             *
+ * *
+ * You should have received a copy of the GNU Affero General Public License        *
+ * along with Holocore.  If not, see <http:></http:>//www.gnu.org/licenses/>.               *
+ */
+package com.projectswg.holocore.services.support.global.network
+
+import com.projectswg.common.network.NetBuffer
+import com.projectswg.common.network.packets.swg.holo.HoloConnectionStopped.ConnectionStoppedReason
+import com.projectswg.holocore.ProjectSWG
+import com.projectswg.holocore.ProjectSWG.CoreException
+import com.projectswg.holocore.intents.support.global.network.CloseConnectionIntent
+import com.projectswg.holocore.intents.support.global.network.ConnectionClosedIntent
+import com.projectswg.holocore.resources.support.data.server_info.mongodb.PswgDatabase
+import com.projectswg.holocore.resources.support.global.network.NetworkClient
+import com.projectswg.holocore.resources.support.global.network.UDPServer
+import com.projectswg.holocore.resources.support.global.network.UDPServer.UDPPacket
+import me.joshlarson.jlcommon.control.IntentHandler
+import me.joshlarson.jlcommon.control.Service
+import me.joshlarson.jlcommon.log.Log
+import java.io.IOException
+import java.net.InetAddress
+import java.net.InetSocketAddress
+import java.nio.ByteBuffer
+import java.nio.channels.AsynchronousServerSocketChannel
+import java.nio.channels.AsynchronousSocketChannel
+import java.nio.channels.CompletionHandler
+import java.util.concurrent.ConcurrentHashMap
+
+class NetworkClientService : Service() {
+	
+	private var tcpServer: AsynchronousServerSocketChannel
+	private val clients: MutableMap<Long, NetworkClient>
+	private val inboundBuffer: ByteBuffer
+	private var udpServer: UDPServer
+	@Volatile
+	private var operational: Boolean = false
+	
+	private val bindPort: Int
+		get() = PswgDatabase.config!!.getInt(this, "bindPort", 44463)
+	
+	init {
+		this.clients = ConcurrentHashMap()
+		this.inboundBuffer = ByteBuffer.allocate(INBOUND_BUFFER_SIZE)
+		this.operational = true
+		
+		val bindPort = bindPort
+		try {
+			tcpServer = AsynchronousServerSocketChannel.open()
+			tcpServer.bind(InetSocketAddress(bindPort), 64)
+			udpServer = UDPServer(bindPort, 32)
+		} catch (e: IOException) {
+			throw CoreException("Failed to start networking", e)
+		}
+		
+		udpServer.setCallback { this.onUdpPacket(it) }
+	}
+	
+	override fun start(): Boolean {
+		tcpServer.accept(null, object : CompletionHandler<AsynchronousSocketChannel, Any?> {
+			override fun completed(result: AsynchronousSocketChannel?, attachment: Any?) {
+				tcpServer.accept(null, this) // starts the next listen
+				if (result != null)
+					handleConnection(result)
+			}
+			
+			override fun failed(exc: Throwable?, attachment: Any?) {
+				if (exc != null)
+					Log.w(exc)
+			}
+		})
+		return true
+	}
+	
+	override fun isOperational(): Boolean {
+		return operational
+	}
+	
+	override fun stop(): Boolean {
+		for (client in clients.values)
+			client.close(ConnectionStoppedReason.APPLICATION)
+		
+		try {
+			tcpServer.close()
+		} catch (e: IOException) {
+			Log.w("Failed to close TCP server")
+			return false
+		}
+		return true
+	}
+	
+	override fun terminate(): Boolean {
+		udpServer.close()
+		return super.terminate()
+	}
+	
+	private fun handleConnection(socket: AsynchronousSocketChannel) {
+		val networkClient = NetworkClient(socket)
+		clients[networkClient.id] = networkClient
+	}
+	
+	private fun disconnect(networkId: Long) {
+		disconnect(clients[networkId])
+	}
+	
+	private fun disconnect(client: NetworkClient?) {
+		if (client == null)
+			return
+		
+		client.close(ConnectionStoppedReason.APPLICATION)
+	}
+	
+	private fun onUdpPacket(packet: UDPPacket) {
+		if (packet.length <= 0)
+			return
+		if (packet.data[0].toInt() == 1) {
+			sendState(packet.address, packet.port)
+		}
+	}
+	
+	private fun sendState(addr: InetAddress, port: Int) {
+		val status = ProjectSWG.getGalaxy().status.name
+		val data = NetBuffer.allocate(3 + status.length)
+		data.addByte(1)
+		data.addAscii(status)
+		udpServer.send(port, addr, data.array())
+	}
+	
+	@IntentHandler
+	private fun handleCloseConnectionIntent(ccii: CloseConnectionIntent) {
+		disconnect(ccii.player.networkId)
+	}
+	
+	@IntentHandler
+	private fun handleConnectionClosedIntent(cci: ConnectionClosedIntent) {
+		disconnect(cci.player.networkId)
+	}
+	
+	companion object {
+		
+		private val INBOUND_BUFFER_SIZE = 4096
+	}
+	
+}