diff --git a/build.gradle b/build.gradle index cfc4e155fb..c1b7b7f497 100644 --- a/build.gradle +++ b/build.gradle @@ -61,6 +61,8 @@ project(":html") { compile "com.badlogicgames.gdx:gdx-controllers:$gdxVersion:sources" compile "com.badlogicgames.gdx:gdx-controllers-gwt:$gdxVersion" compile "com.badlogicgames.gdx:gdx-controllers-gwt:$gdxVersion:sources" + compile "com.sksamuel.gwt:gwt-websockets:1.0.4" + compile "com.sksamuel.gwt:gwt-websockets:1.0.4:sources" } } @@ -116,6 +118,7 @@ project(":kryonet") { dependencies { compile project(":core") compile 'com.github.crykn:kryonet:2.22.1' + compile "org.java-websocket:Java-WebSocket:1.3.7" } } diff --git a/core/src/io/anuke/mindustry/Mindustry.java b/core/src/io/anuke/mindustry/Mindustry.java index c6dbd7f055..e844d43ec3 100644 --- a/core/src/io/anuke/mindustry/Mindustry.java +++ b/core/src/io/anuke/mindustry/Mindustry.java @@ -38,6 +38,7 @@ public class Mindustry extends ModuleCore { @Override public void dispose() { platforms.onGameExit(); + Net.dispose(); super.dispose(); } diff --git a/core/src/io/anuke/mindustry/Vars.java b/core/src/io/anuke/mindustry/Vars.java index 89b8f38e3a..ce43ca4485 100644 --- a/core/src/io/anuke/mindustry/Vars.java +++ b/core/src/io/anuke/mindustry/Vars.java @@ -75,6 +75,7 @@ public class Vars{ //server port public static final int port = 6567; + public static final int webPort = 6568; public static Control control; public static Renderer renderer; diff --git a/core/src/io/anuke/mindustry/core/NetServer.java b/core/src/io/anuke/mindustry/core/NetServer.java index b3274e7828..218fb0fa5e 100644 --- a/core/src/io/anuke/mindustry/core/NetServer.java +++ b/core/src/io/anuke/mindustry/core/NetServer.java @@ -9,6 +9,7 @@ import io.anuke.mindustry.entities.Player; import io.anuke.mindustry.entities.SyncEntity; import io.anuke.mindustry.entities.TileEntity; import io.anuke.mindustry.entities.enemies.Enemy; +import io.anuke.mindustry.net.NetConnection; import io.anuke.mindustry.net.NetworkIO; import io.anuke.mindustry.net.Net; import io.anuke.mindustry.net.Net.SendMode; @@ -370,10 +371,10 @@ public class NetServer extends Module{ if(Timers.get("serverBlockSync", blockSyncTime)){ - IntArray connections = Net.getConnections(); + Array connections = new Array<>(); for(int i = 0; i < connections.size; i ++){ - int id = connections.get(i); + int id = connections.get(i).id; Player player = this.connections.get(id); if(player == null) continue; int x = Mathf.scl2(player.x, Vars.tilesize); diff --git a/core/src/io/anuke/mindustry/graphics/BlockRenderer.java b/core/src/io/anuke/mindustry/graphics/BlockRenderer.java index 332ae2030b..5540e72dde 100644 --- a/core/src/io/anuke/mindustry/graphics/BlockRenderer.java +++ b/core/src/io/anuke/mindustry/graphics/BlockRenderer.java @@ -1,21 +1,17 @@ package io.anuke.mindustry.graphics; -import static io.anuke.mindustry.Vars.*; -import static io.anuke.ucore.core.Core.camera; - -import java.util.Arrays; - import com.badlogic.gdx.Gdx; import com.badlogic.gdx.graphics.Color; import com.badlogic.gdx.graphics.GL20; import com.badlogic.gdx.graphics.OrthographicCamera; import com.badlogic.gdx.utils.Array; - import io.anuke.mindustry.Vars; import io.anuke.mindustry.core.GameState; import io.anuke.mindustry.core.GameState.State; import io.anuke.mindustry.game.SpawnPoint; -import io.anuke.mindustry.world.*; +import io.anuke.mindustry.world.Block; +import io.anuke.mindustry.world.Layer; +import io.anuke.mindustry.world.Tile; import io.anuke.mindustry.world.blocks.Blocks; import io.anuke.mindustry.world.blocks.types.StaticBlock; import io.anuke.ucore.core.Core; @@ -24,6 +20,11 @@ import io.anuke.ucore.core.Graphics; import io.anuke.ucore.graphics.CacheBatch; import io.anuke.ucore.util.Mathf; +import java.util.Arrays; + +import static io.anuke.mindustry.Vars.*; +import static io.anuke.ucore.core.Core.camera; + public class BlockRenderer{ private final static int chunksize = 32; private final static int initialRequests = 32*32; @@ -269,6 +270,6 @@ public class BlockRenderer{ cache = null; if(cbatch != null) cbatch.dispose(); - cbatch = new CacheBatch(Vars.world.width() * Vars.world.height() * 3); + cbatch = new CacheBatch(Vars.world.width() * Vars.world.height() * 4); } } diff --git a/core/src/io/anuke/mindustry/input/DesktopInput.java b/core/src/io/anuke/mindustry/input/DesktopInput.java index 8e2f09d1d5..2d42e5f80e 100644 --- a/core/src/io/anuke/mindustry/input/DesktopInput.java +++ b/core/src/io/anuke/mindustry/input/DesktopInput.java @@ -92,7 +92,7 @@ public class DesktopInput extends InputHandler{ for(int i = 1; i <= 6 && i <= control.getWeapons().size; i ++){ if(Inputs.keyTap("weapon_" + i)){ player.weaponLeft = player.weaponRight = control.getWeapons().get(i - 1); - Vars.netClient.handleWeaponSwitch(); + if(Net.active()) Vars.netClient.handleWeaponSwitch(); Vars.ui.hudfrag.updateWeapons(); } } diff --git a/core/src/io/anuke/mindustry/net/Net.java b/core/src/io/anuke/mindustry/net/Net.java index 196c6d6868..ee997018ef 100644 --- a/core/src/io/anuke/mindustry/net/Net.java +++ b/core/src/io/anuke/mindustry/net/Net.java @@ -2,7 +2,6 @@ package io.anuke.mindustry.net; import com.badlogic.gdx.Gdx; import com.badlogic.gdx.utils.Array; -import com.badlogic.gdx.utils.IntArray; import com.badlogic.gdx.utils.IntMap; import com.badlogic.gdx.utils.ObjectMap; import com.badlogic.gdx.utils.async.AsyncExecutor; @@ -85,7 +84,7 @@ public class Net{ } /**Returns a list of all connections IDs.*/ - public static IntArray getConnections(){ + public static Array getConnections(){ return serverProvider.getConnections(); } @@ -206,8 +205,8 @@ public class Net{ } public static void dispose(){ - clientProvider.dispose(); - serverProvider.dispose(); + if(clientProvider != null) clientProvider.dispose(); + if(serverProvider != null) serverProvider.dispose(); executor.dispose(); } @@ -218,51 +217,51 @@ public class Net{ } /**Client implementation.*/ - public static interface ClientProvider { + public interface ClientProvider { /**Connect to a server.*/ - public void connect(String ip, int port) throws IOException; + void connect(String ip, int port) throws IOException; /**Send an object to the server.*/ - public void send(Object object, SendMode mode); + void send(Object object, SendMode mode); /**Update the ping. Should be done every second or so.*/ - public void updatePing(); + void updatePing(); /**Get ping in milliseconds. Will only be valid after a call to updatePing.*/ - public int getPing(); + int getPing(); /**Disconnect from the server.*/ - public void disconnect(); + void disconnect(); /**Discover servers. This should block for a certain amount of time, and will most likely be run in a different thread.*/ - public Array discover(); + Array discover(); /**Ping a host. If an error occured, failed() should be called with the exception. */ - public void pingHost(String address, int port, Consumer valid, Consumer failed); + void pingHost(String address, int port, Consumer valid, Consumer failed); /**Register classes to be sent.*/ - public void register(Class... types); + void register(Class... types); /**Close all connections.*/ - public void dispose(); + void dispose(); } /**Server implementation.*/ - public static interface ServerProvider { + public interface ServerProvider { /**Host a server at specified port.*/ - public void host(int port) throws IOException; + void host(int port) throws IOException; /**Sends a large stream of data to a specific client.*/ - public void sendStream(int id, Streamable stream); + void sendStream(int id, Streamable stream); /**Send an object to everyone connected.*/ - public void send(Object object, SendMode mode); + void send(Object object, SendMode mode); /**Send an object to a specific client ID.*/ - public void sendTo(int id, Object object, SendMode mode); + void sendTo(int id, Object object, SendMode mode); /**Send an object to everyone except a client ID.*/ - public void sendExcept(int id, Object object, SendMode mode); + void sendExcept(int id, Object object, SendMode mode); /**Close the server connection.*/ - public void close(); + void close(); /**Return all connected users.*/ - public IntArray getConnections(); + Array getConnections(); /**Kick a certain connection.*/ - public void kick(int connection); + void kick(int connection); /**Returns the ping for a certain connection.*/ - public int getPingFor(int connection); + int getPingFor(NetConnection connection); /**Register classes to be sent.*/ - public void register(Class... types); + void register(Class... types); /**Close all connections.*/ - public void dispose(); + void dispose(); } public enum SendMode{ diff --git a/core/src/io/anuke/mindustry/net/NetConnection.java b/core/src/io/anuke/mindustry/net/NetConnection.java new file mode 100644 index 0000000000..ba6bfbf80c --- /dev/null +++ b/core/src/io/anuke/mindustry/net/NetConnection.java @@ -0,0 +1,16 @@ +package io.anuke.mindustry.net; + +import io.anuke.mindustry.net.Net.SendMode; + +public abstract class NetConnection { + public final int id; + public final String address; + + public NetConnection(int id, String address){ + this.id = id; + this.address = address; + } + + public abstract void send(Object object, SendMode mode); + public abstract void close(); +} diff --git a/desktop/src/io/anuke/mindustry/desktop/DesktopLauncher.java b/desktop/src/io/anuke/mindustry/desktop/DesktopLauncher.java index 4cf614c21f..524f80ad84 100644 --- a/desktop/src/io/anuke/mindustry/desktop/DesktopLauncher.java +++ b/desktop/src/io/anuke/mindustry/desktop/DesktopLauncher.java @@ -6,7 +6,7 @@ import club.minnced.discord.rpc.DiscordRichPresence; import com.badlogic.gdx.backends.lwjgl3.Lwjgl3Application; import com.badlogic.gdx.backends.lwjgl3.Lwjgl3ApplicationConfiguration; import com.badlogic.gdx.utils.Array; -import io.anuke.kryonet.KryoClient; +import io.anuke.kryonet.JavaWebsocketClient; import io.anuke.kryonet.KryoServer; import io.anuke.mindustry.Mindustry; import io.anuke.mindustry.Vars; @@ -120,7 +120,8 @@ public class DesktopLauncher { Mindustry.args = Array.with(arg); - Net.setClientProvider(new KryoClient()); + //Net.setClientProvider(new KryoClient()); + Net.setClientProvider(new JavaWebsocketClient()); Net.setServerProvider(new KryoServer()); try { diff --git a/html/src/io/anuke/mindustry/GdxDefinition.gwt.xml b/html/src/io/anuke/mindustry/GdxDefinition.gwt.xml index f29f12e49e..f7ce42f271 100644 --- a/html/src/io/anuke/mindustry/GdxDefinition.gwt.xml +++ b/html/src/io/anuke/mindustry/GdxDefinition.gwt.xml @@ -6,6 +6,7 @@ + diff --git a/html/src/io/anuke/mindustry/client/HtmlLauncher.java b/html/src/io/anuke/mindustry/client/HtmlLauncher.java index 808dc291cf..9209ab2c88 100644 --- a/html/src/io/anuke/mindustry/client/HtmlLauncher.java +++ b/html/src/io/anuke/mindustry/client/HtmlLauncher.java @@ -17,6 +17,7 @@ import com.google.gwt.user.client.Window; import com.google.gwt.user.client.ui.*; import io.anuke.mindustry.Mindustry; import io.anuke.mindustry.io.PlatformFunction; +import io.anuke.mindustry.net.Net; import java.util.Date; @@ -90,6 +91,8 @@ public class HtmlLauncher extends GwtApplication { setupResizeHook(); } }); + + Net.setClientProvider(new WebsocketClient()); Mindustry.platforms = new PlatformFunction(){ DateTimeFormat format = DateTimeFormat.getFormat("EEE, dd MMM yyyy HH:mm:ss"); diff --git a/html/src/io/anuke/mindustry/client/WebsocketClient.java b/html/src/io/anuke/mindustry/client/WebsocketClient.java new file mode 100644 index 0000000000..2885c4bacc --- /dev/null +++ b/html/src/io/anuke/mindustry/client/WebsocketClient.java @@ -0,0 +1,115 @@ +package io.anuke.mindustry.client; + +import com.badlogic.gdx.utils.Array; +import com.badlogic.gdx.utils.Base64Coder; +import com.badlogic.gdx.utils.reflect.ClassReflection; +import com.badlogic.gdx.utils.reflect.ReflectionException; +import com.sksamuel.gwt.websockets.Websocket; +import com.sksamuel.gwt.websockets.WebsocketListener; +import io.anuke.mindustry.Vars; +import io.anuke.mindustry.net.Host; +import io.anuke.mindustry.net.Net; +import io.anuke.mindustry.net.Net.ClientProvider; +import io.anuke.mindustry.net.Net.SendMode; +import io.anuke.mindustry.net.Packet; +import io.anuke.mindustry.net.Packets.Connect; +import io.anuke.mindustry.net.Packets.Disconnect; +import io.anuke.mindustry.net.Registrator; +import io.anuke.ucore.function.Consumer; + +import java.io.IOException; +import java.nio.ByteBuffer; + +public class WebsocketClient implements ClientProvider { + Websocket socket; + ByteBuffer buffer = ByteBuffer.allocate(1024); + + @Override + public void connect(String ip, int port) throws IOException { + socket = new Websocket("ws://" + ip + ":" + Vars.webPort); + socket.addListener(new WebsocketListener() { + public void onMessage(byte[] bytes) { + try { + ByteBuffer buffer = ByteBuffer.wrap(bytes); + byte id = buffer.get(); + if(id == -2){ + //this is a framework message... do nothing yet? + }else { + Class type = Registrator.getByID(id); + Packet packet = (Packet) ClassReflection.newInstance(type); + packet.read(buffer); + Net.handleClientReceived(packet); + } + }catch (ReflectionException e){ + throw new RuntimeException(e); + } + } + + @Override + public void onClose() { + Disconnect disconnect = new Disconnect(); + Net.handleClientReceived(disconnect); + } + + @Override + public void onMessage(String msg) { + onMessage(Base64Coder.decode(msg)); + } + + @Override + public void onOpen() { + Connect connect = new Connect(); + Net.handleClientReceived(connect); + } + }); + socket.open(); + } + + @Override + public void send(Object object, SendMode mode) { + if(!(object instanceof Packet)) throw new RuntimeException("All sent objects must be packets!"); + Packet p = (Packet)object; + buffer.position(0); + buffer.put(Registrator.getID(object.getClass())); + p.write(buffer); + int pos = buffer.position(); + buffer.position(0); + byte[] out = new byte[pos]; + buffer.get(out); + String string = new String(Base64Coder.encode(out)); + socket.send(string); + } + + @Override + public void updatePing() { + + } + + @Override + public int getPing() { + return 0; + } + + @Override + public void disconnect() { + socket.close(); + } + + @Override + public Array discover() { + return new Array<>(); + } + + @Override + public void pingHost(String address, int port, Consumer valid, Consumer failed) { + failed.accept(new IOException()); + } + + @Override + public void register(Class... types) { } + + @Override + public void dispose() { + socket.close(); + } +} diff --git a/kryonet/src/io/anuke/kryonet/JavaWebsocketClient.java b/kryonet/src/io/anuke/kryonet/JavaWebsocketClient.java new file mode 100644 index 0000000000..b053cdf144 --- /dev/null +++ b/kryonet/src/io/anuke/kryonet/JavaWebsocketClient.java @@ -0,0 +1,147 @@ +package io.anuke.kryonet; + +import com.badlogic.gdx.utils.Array; +import com.badlogic.gdx.utils.Base64Coder; +import com.badlogic.gdx.utils.reflect.ClassReflection; +import io.anuke.mindustry.Vars; +import io.anuke.mindustry.net.Host; +import io.anuke.mindustry.net.Net; +import io.anuke.mindustry.net.Net.ClientProvider; +import io.anuke.mindustry.net.Net.SendMode; +import io.anuke.mindustry.net.Packet; +import io.anuke.mindustry.net.Packets.Connect; +import io.anuke.mindustry.net.Packets.Disconnect; +import io.anuke.mindustry.net.Registrator; +import io.anuke.ucore.UCore; +import io.anuke.ucore.function.Consumer; +import org.java_websocket.client.WebSocketClient; +import org.java_websocket.drafts.Draft_6455; +import org.java_websocket.handshake.ServerHandshake; + +import java.io.IOException; +import java.net.URI; +import java.net.URISyntaxException; +import java.nio.ByteBuffer; + +public class JavaWebsocketClient implements ClientProvider { + WebSocketClient socket; + ByteBuffer buffer = ByteBuffer.allocate(1024); + boolean debug = false; + + @Override + public void connect(String ip, int port) throws IOException { + try { + URI i = new URI("ws://" + ip + ":" + Vars.webPort); + UCore.log("Connecting: " + i); + socket = new WebSocketClient(i, new Draft_6455(), null, 5000) { + Thread thread; + + @Override + public void connect() { + if(thread != null ) + throw new IllegalStateException( "WebSocketClient objects are not reuseable" ); + thread = new Thread(this); + thread.setDaemon(true); + thread.start(); + } + + @Override + public void onOpen(ServerHandshake handshakedata) { + UCore.log("Connected!"); + Connect connect = new Connect(); + Net.handleClientReceived(connect); + } + + @Override + public void onMessage(String message) { + if(debug) UCore.log("Got message: " + message); + try { + byte[] bytes = Base64Coder.decode(message); + ByteBuffer buffer = ByteBuffer.wrap(bytes); + byte id = buffer.get(); + if (id == -2) { + //this is a framework message... do nothing yet? + } else { + Class type = Registrator.getByID(id); + if(debug) UCore.log("Got class ID: " + type); + Packet packet = (Packet) ClassReflection.newInstance(type); + packet.read(buffer); + Net.handleClientReceived(packet); + } + } catch (Exception e) { + e.printStackTrace(); + //throw new RuntimeException(e); + } + } + + @Override + public void onClose(int code, String reason, boolean remote) { + if(debug) UCore.log("Closed."); + Disconnect disconnect = new Disconnect(); + Net.handleClientReceived(disconnect); + } + + @Override + public void onError(Exception ex) { + onClose(0, null, true); + ex.printStackTrace(); + } + }; + socket.connect(); + }catch (URISyntaxException e){ + throw new IOException(e); + } + } + + @Override + public void send(Object object, SendMode mode) { + if(!(object instanceof Packet)) throw new RuntimeException("All sent objects must be packets!"); + Packet p = (Packet)object; + buffer.position(0); + buffer.put(Registrator.getID(object.getClass())); + p.write(buffer); + int pos = buffer.position(); + buffer.position(0); + byte[] out = new byte[pos]; + buffer.get(out); + String string = new String(Base64Coder.encode(out)); + if(debug) UCore.log("Sending string: " + string); + socket.send(string); + } + + @Override + public void updatePing() { + + } + + @Override + public int getPing() { + return 0; + } + + @Override + public void disconnect() { + socket.close(); + } + + @Override + public Array discover() { + return new Array<>(); + } + + @Override + public void pingHost(String address, int port, Consumer valid, Consumer failed) { + failed.accept(new IOException()); + } + + @Override + public void register(Class... types) { } + + @Override + public void dispose() { + if(socket != null) socket.close(); + for(Thread thread : Thread.getAllStackTraces().keySet()){ + if(thread.getName().equals("WebsocketWriteThread")) thread.interrupt(); + } + } +} diff --git a/kryonet/src/io/anuke/kryonet/KryoServer.java b/kryonet/src/io/anuke/kryonet/KryoServer.java index a507b7f081..a939422a40 100644 --- a/kryonet/src/io/anuke/kryonet/KryoServer.java +++ b/kryonet/src/io/anuke/kryonet/KryoServer.java @@ -1,16 +1,19 @@ package io.anuke.kryonet; import com.badlogic.gdx.Gdx; -import com.badlogic.gdx.utils.IntArray; +import com.badlogic.gdx.utils.Array; +import com.badlogic.gdx.utils.Base64Coder; import com.esotericsoftware.kryonet.Connection; import com.esotericsoftware.kryonet.FrameworkMessage; import com.esotericsoftware.kryonet.Listener; import com.esotericsoftware.kryonet.Listener.LagListener; import com.esotericsoftware.kryonet.Server; import com.esotericsoftware.kryonet.util.InputStreamSender; +import io.anuke.mindustry.Vars; import io.anuke.mindustry.net.Net; import io.anuke.mindustry.net.Net.SendMode; import io.anuke.mindustry.net.Net.ServerProvider; +import io.anuke.mindustry.net.NetConnection; import io.anuke.mindustry.net.Packets.Connect; import io.anuke.mindustry.net.Packets.Disconnect; import io.anuke.mindustry.net.Packets.KickPacket; @@ -21,13 +24,25 @@ import io.anuke.mindustry.net.Streamable.StreamBegin; import io.anuke.mindustry.net.Streamable.StreamChunk; import io.anuke.ucore.UCore; import io.anuke.ucore.core.Timers; +import org.java_websocket.WebSocket; +import org.java_websocket.handshake.ClientHandshake; +import org.java_websocket.server.WebSocketServer; import java.io.IOException; +import java.net.InetSocketAddress; import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.concurrent.CopyOnWriteArrayList; public class KryoServer implements ServerProvider { - Server server; - IntArray connections = new IntArray(); + final Server server; + final SocketServer webServer; + final ByteSerializer serializer = new ByteSerializer(); + final ByteBuffer buffer = ByteBuffer.allocate(4096); + final CopyOnWriteArrayList connections = new CopyOnWriteArrayList<>(); + final Array array = new Array<>(); + + int lastconnection = 0; public KryoServer(){ server = new Server(4096*2, 2048, connection -> new ByteSerializer()); //TODO tweak @@ -38,18 +53,21 @@ public class KryoServer implements ServerProvider { datagramChannel.send(buffer, fromAddress); return true; }); + webServer = new SocketServer(Vars.webPort); Listener listener = new Listener(){ @Override public void connected (Connection connection) { + KryoConnection kn = new KryoConnection(lastconnection ++, connection.getRemoteAddressTCP().toString(), connection); + Connect c = new Connect(); - c.id = connection.getID(); + c.id = kn.id; c.addressTCP = connection.getRemoteAddressTCP().toString(); try { - Net.handleServerReceived(c, c.id); - connections.add(c.id); + Net.handleServerReceived(c, kn.id); + connections.add(kn); }catch (Exception e){ Gdx.app.postRunnable(() -> {throw new RuntimeException(e);}); } @@ -57,10 +75,12 @@ public class KryoServer implements ServerProvider { @Override public void disconnected (Connection connection) { - connections.removeValue(connection.getID()); + KryoConnection k = getByKryoID(connection.getID()); + if(k == null) return; + connections.remove(k); Disconnect c = new Disconnect(); - c.id = connection.getID(); + c.id = k.id; try{ Net.handleServerReceived(c, c.id); @@ -71,14 +91,13 @@ public class KryoServer implements ServerProvider { @Override public void received (Connection connection, Object object) { - if(object instanceof FrameworkMessage) return; + KryoConnection k = getByKryoID(connection.getID()); + if(object instanceof FrameworkMessage || k == null) return; try{ - Net.handleServerReceived(object, connection.getID()); + Net.handleServerReceived(object, k.id); }catch (Exception e){ - //...do absolutely nothing. e.printStackTrace(); - //Gdx.app.postRunnable(() -> {throw new RuntimeException(e);}); } } }; @@ -93,33 +112,30 @@ public class KryoServer implements ServerProvider { } @Override - public IntArray getConnections() { - return connections; + public Array getConnections() { + array.clear(); + for(KryoConnection c : connections){ + array.add(c); + } + return array; } @Override public void kick(int connection) { - Connection conn = getByID(connection); - - if(conn == null){ - connections.removeValue(connection); - return; - } + KryoConnection con = getByID(connection); KickPacket p = new KickPacket(); p.reason = KickReason.kick; - conn.sendTCP(p); - Timers.runTask(1f, () -> { - if(conn.isConnected()){ - conn.close(); - } - }); + con.send(p, SendMode.tcp); + Timers.runTask(1f, con::close); } @Override public void host(int port) throws IOException { + lastconnection = 0; server.bind(port, port); + webServer.start(); Thread thread = new Thread(() -> { try{ @@ -136,65 +152,91 @@ public class KryoServer implements ServerProvider { public void close() { UCore.setPrivate(server, "shutdown", true); - new Thread(() -> server.close()).run(); + new Thread(() ->{ + try { + server.close(); + webServer.stop(); + }catch (Exception e){ + Gdx.app.postRunnable(() -> {throw new RuntimeException(e);}); + } + }).run(); } @Override public void sendStream(int id, Streamable stream) { - Connection connection = getByID(id); + KryoConnection connection = getByID(id); if(connection == null) return; + try { - connection.addListener(new InputStreamSender(stream.stream, 512) { - int id; + if (connection.connection != null) { - protected void start () { - //send an object so the receiving side knows how to handle the following chunks + connection.connection.addListener(new InputStreamSender(stream.stream, 512) { + int id; + + protected void start() { + //send an object so the receiving side knows how to handle the following chunks + StreamBegin begin = new StreamBegin(); + begin.total = stream.stream.available(); + begin.type = stream.getClass(); + connection.connection.sendTCP(begin); + id = begin.id; + } + + protected Object next(byte[] bytes) { + StreamChunk chunk = new StreamChunk(); + chunk.id = id; + chunk.data = bytes; + return chunk; //wrap the byte[] with an object so the receiving side knows how to handle it. + } + }); + } else { + int cid; StreamBegin begin = new StreamBegin(); begin.total = stream.stream.available(); begin.type = stream.getClass(); - connection.sendTCP(begin); - id = begin.id; - } + connection.send(begin, SendMode.tcp); + cid = begin.id; - protected Object next (byte[] bytes) { - StreamChunk chunk = new StreamChunk(); - chunk.id = id; - chunk.data = bytes; - return chunk; //wrap the byte[] with an object so the receiving side knows how to handle it. + while (stream.stream.available() > 0) { + byte[] bytes = new byte[Math.min(512, stream.stream.available())]; + stream.stream.read(bytes); + + StreamChunk chunk = new StreamChunk(); + chunk.id = cid; + chunk.data = bytes; + connection.send(chunk, SendMode.tcp); + } } - }); + }catch (IOException e){ + throw new RuntimeException(e); + } } @Override public void send(Object object, SendMode mode) { - if(mode == SendMode.tcp){ - server.sendToAllTCP(object); - }else{ - server.sendToAllUDP(object); + for(int i = 0; i < connections.size(); i ++){ + connections.get(i).send(object, mode); } } @Override public void sendTo(int id, Object object, SendMode mode) { - if(mode == SendMode.tcp){ - server.sendToTCP(id, object); - }else{ - server.sendToUDP(id, object); - } + NetConnection conn = getByID(id); + conn.send(object, mode); } @Override public void sendExcept(int id, Object object, SendMode mode) { - if(mode == SendMode.tcp){ - server.sendToAllExceptTCP(id, object); - }else{ - server.sendToAllExceptUDP(id, object); + for(int i = 0; i < connections.size(); i ++){ + KryoConnection conn = connections.get(i); + if(conn.id != id) conn.send(object, mode); } } @Override - public int getPingFor(int connection) { - return getByID(connection).getReturnTripTime(); + public int getPingFor(NetConnection con) { + KryoConnection k = (KryoConnection)con; + return k.connection == null ? 0 : k.connection.getReturnTripTime(); } @Override @@ -204,7 +246,8 @@ public class KryoServer implements ServerProvider { public void dispose(){ try { server.dispose(); - }catch (IOException e){ + webServer.stop(); + }catch (Exception e){ throw new RuntimeException(e); } } @@ -213,13 +256,146 @@ public class KryoServer implements ServerProvider { Gdx.app.postRunnable(() -> { throw new RuntimeException(e);}); } - Connection getByID(int id){ - for(Connection con : server.getConnections()){ - if(con.getID() == id){ + KryoConnection getByID(int id){ + for(int i = 0; i < connections.size(); i ++){ + KryoConnection con = connections.get(i); + if(con.id == id){ return con; } } return null; } + + KryoConnection getByKryoID(int id){ + for(int i = 0; i < connections.size(); i ++){ + KryoConnection con = connections.get(i); + if(con.connection != null && con.connection.getID() == id){ + return con; + } + } + + return null; + } + + KryoConnection getBySocket(WebSocket socket){ + for(int i = 0; i < connections.size(); i ++){ + KryoConnection con = connections.get(i); + if(con.socket == socket){ + return con; + } + } + + return null; + } + + class KryoConnection extends NetConnection{ + public final WebSocket socket; + public final Connection connection; + + public KryoConnection(int id, String address, WebSocket socket) { + super(id, address); + this.socket = socket; + this.connection = null; + } + + public KryoConnection(int id, String address, Connection connection) { + super(id, address); + this.socket = null; + this.connection = connection; + } + + @Override + public void send(Object object, SendMode mode){ + if(socket != null){ + try { + synchronized (buffer) { + buffer.position(0); + UCore.log("Sending object with ID " + Registrator.getID(object.getClass())); + serializer.write(buffer, object); + int pos = buffer.position(); + buffer.position(0); + byte[] out = new byte[pos]; + buffer.get(out); + String string = new String(Base64Coder.encode(out)); + UCore.log("Sending string: " + string); + socket.send(string); + } + }catch (Exception e){ + e.printStackTrace(); + connections.remove(this); + } + }else if (connection != null) { + if (mode == SendMode.tcp) { + connection.sendTCP(object); + } else { + connection.sendUDP(object); + } + } + } + + @Override + public void close(){ + if(socket != null){ + if(socket.isOpen()) socket.close(); + }else if (connection != null) { + if(connection.isConnected()) connection.close(); + } + } + + } + + class SocketServer extends WebSocketServer { + + public SocketServer(int port) { + super(new InetSocketAddress(port)); + } + + @Override + public void onOpen(WebSocket conn, ClientHandshake handshake) { + Connect connect = new Connect(); + connect.addressTCP = conn.getRemoteSocketAddress().toString(); + UCore.log("Websocket connection recieved: " + connect.addressTCP); + KryoConnection kn = new KryoConnection(lastconnection ++, connect.addressTCP, conn); + connections.add(kn); + } + + @Override + public void onClose(WebSocket conn, int code, String reason, boolean remote) { + if (conn == null) return; + Disconnect disconnect = new Disconnect(); + KryoConnection k = getBySocket(conn); + if(k != null) Net.handleServerReceived(disconnect, k.id); + } + + @Override + public void onMessage(WebSocket conn, String message) { + try { + UCore.log("Got message: " + message); + KryoConnection k = getBySocket(conn); + if (k == null) return; + + byte[] out = Base64Coder.decode(message); + UCore.log("Decoded: " + Arrays.toString(out)); + ByteBuffer buffer = ByteBuffer.wrap(out); + Object o = serializer.read(buffer); + Net.handleServerReceived(o, k.id); + }catch (Exception e){ + UCore.log("Error reading message!"); + e.printStackTrace(); + } + } + + @Override + public void onError(WebSocket conn, Exception ex) { + UCore.log("WS error:"); + ex.printStackTrace(); + } + + @Override + public void onStart() { + UCore.log("Web server started."); + } + } + }