Experimental MNet-2 backend

net/src/io/anuke/mindustry/net/ArcNetClient.java

@@ -8,7 +8,6 @@ import io.anuke.arc.util.async.*;
 import io.anuke.arc.util.pooling.*;
 import io.anuke.mindustry.net.Net.*;
 import io.anuke.mindustry.net.Packets.*;
-import net.jpountz.lz4.*;
 
 import java.io.*;
 import java.net.*;
@@ -20,7 +19,6 @@ import static io.anuke.mindustry.Vars.*;
 public class ArcNetClient implements ClientProvider{
     final Client client;
     final Supplier<DatagramPacket> packetSupplier = () -> new DatagramPacket(new byte[256], 256);
-    final LZ4FastDecompressor decompressor = LZ4Factory.fastestInstance().fastDecompressor();
 
     public ArcNetClient(){
         client = new Client(8192, 4096, new PacketSerializer());
@@ -75,14 +73,6 @@ public class ArcNetClient implements ClientProvider{
         }
     }
 
-
-    @Override
-    public byte[] decompressSnapshot(byte[] input, int size){
-        byte[] result = new byte[size];
-        decompressor.decompress(input, result);
-        return result;
-    }
-
     @Override
     public void connect(String ip, int port, Runnable success){
         Threads.daemon(() -> {

net/src/io/anuke/mindustry/net/ArcNetServer.java

@@ -6,7 +6,6 @@ import io.anuke.arc.util.*;
 import io.anuke.arc.util.async.*;
 import io.anuke.mindustry.net.Net.*;
 import io.anuke.mindustry.net.Packets.*;
-import net.jpountz.lz4.*;
 
 import java.io.*;
 import java.nio.*;
@@ -18,7 +17,6 @@ import static io.anuke.mindustry.Vars.*;
 public class ArcNetServer implements ServerProvider{
     final Server server;
     final CopyOnWriteArrayList<ArcConnection> connections = new CopyOnWriteArrayList<>();
-    final LZ4Compressor compressor = LZ4Factory.fastestInstance().fastCompressor();
     Thread serverThread;
 
     public ArcNetServer(){
@@ -87,11 +85,6 @@ public class ArcNetServer implements ServerProvider{
         server.addListener(listener);
     }
 
-    @Override
-    public byte[] compressSnapshot(byte[] input){
-        return compressor.compress(input);
-    }
-
     @Override
     public Iterable<ArcConnection> getConnections(){
         return connections;

net/src/io/anuke/mindustry/net/MClient.java

@@ -0,0 +1,177 @@
+package io.anuke.mindustry.net;
+
+import io.anuke.arc.*;
+import io.anuke.arc.function.*;
+import io.anuke.arc.util.async.*;
+import io.anuke.mindustry.game.EventType.*;
+import io.anuke.mindustry.net.Net.*;
+import io.anuke.mindustry.net.Packets.*;
+import io.anuke.mnet.*;
+import io.anuke.mnet.MSocket;
+import io.anuke.mnet.MSocketImpl;
+
+import java.io.*;
+import java.net.*;
+import java.nio.*;
+import java.util.*;
+
+import static io.anuke.mindustry.Vars.*;
+
+public class MClient implements ClientProvider, ApplicationListener{
+    MSocket socket;
+
+    public MClient(){
+        Events.on(AppLoadEvent.class, e -> {
+            Core.app.addListener(this);
+        });
+    }
+
+    public void connect(String ip, int port, Runnable success) throws IOException{
+        socket = new MSocketImpl(InetAddress.getByName(ip), port, new PacketSerializer());
+        socket.addDcListener((sock, reason) -> Core.app.post(() -> Net.handleClientReceived(new Disconnect())));
+        socket.connectAsync(null, 2000, response -> {
+            if(response.getType() == ResponseType.ACCEPTED){
+                Core.app.post(() -> {
+                    success.run();
+                    Net.handleClientReceived(new Connect());
+                });
+            }else if(response.getType() == ResponseType.WRONG_STATE){
+                Core.app.post(() -> Net.showError(new IOException("alreadyconnected")));
+            }else{
+                Core.app.post(() -> Net.showError(new IOException("connection refused")));
+            }
+        });
+    }
+
+    @Override
+    public void updatePing(){
+
+    }
+
+    @Override
+    public void dispose(){
+        disconnect();
+    }
+
+    public void send(Object object, SendMode mode){
+        if(mode == SendMode.tcp){
+            socket.send(object);
+        }else{
+            socket.sendUnreliable(object);
+        }
+    }
+
+    public void update(){
+        if(socket == null) return;
+
+        socket.update((sock, object) -> Core.app.post(() -> {
+            try{
+                Net.handleClientReceived(object);
+            }catch(Exception e){
+                Net.showError(e);
+                netClient.disconnectQuietly();
+            }
+        }));
+    }
+
+    public int getPing(){
+        return socket == null ? 0 : (int)socket.getPing();
+    }
+
+    public void disconnect(){
+        if(socket != null) socket.close();
+    }
+
+    public void discover(Consumer<Host> callback, Runnable done){
+        Threads.daemon(() -> {
+            byte[] bytes = new byte[512];
+            ByteBuffer buffer = ByteBuffer.wrap(bytes);
+            DatagramPacket packet = new DatagramPacket(bytes, bytes.length);
+            ArrayList<InetAddress> foundAddresses = new ArrayList<>();
+
+            try(DatagramSocket socket = new DatagramSocket()){
+                broadcast(port, socket);
+
+                socket.setSoTimeout(4000);
+
+                outer:
+                while(true){
+
+                    try{
+                        socket.receive(packet);
+                    }catch(SocketTimeoutException ex){
+                        done.run();
+                        return;
+                    }
+
+                    buffer.position(0);
+
+                    InetAddress address = ((InetSocketAddress)packet.getSocketAddress()).getAddress();
+
+                    for(InetAddress other : foundAddresses){
+                        if(other.equals(address) || (isLocal(other) && isLocal(address))){
+                            continue outer;
+                        }
+                    }
+
+                    Host host = NetworkIO.readServerData(address.getHostName(), buffer);
+                    callback.accept(host);
+                    foundAddresses.add(address);
+                }
+            }catch(IOException ex){
+                done.run();
+            }
+        });
+    }
+
+    public void pingHost(String address, int port, Consumer<Host> valid, Consumer<Exception> failed){
+        Threads.daemon(() -> {
+            try{
+                DatagramPacket packet = new DatagramPacket(new byte[512], 512);
+
+                DatagramSocket socket = new DatagramSocket();
+                socket.send(new DatagramPacket(new byte[]{-2}, 1, InetAddress.getByName(address), port));
+                socket.setSoTimeout(4000);
+                socket.receive(packet);
+
+                ByteBuffer buffer = ByteBuffer.wrap(packet.getData());
+                Host host = NetworkIO.readServerData(packet.getAddress().getHostAddress(), buffer);
+
+                Core.app.post(() -> valid.accept(host));
+            }catch(Exception e){
+                Core.app.post(() -> failed.accept(e));
+            }
+        });
+    }
+    private void broadcast (int udpPort, DatagramSocket socket) throws IOException{
+        byte[] data = {-2};
+
+        for (NetworkInterface iface : Collections.list(NetworkInterface.getNetworkInterfaces())){
+            for (InetAddress address : Collections.list(iface.getInetAddresses())){
+
+                byte[] ip = address.getAddress(); //255.255.255.255
+                try{
+                    socket.send(new DatagramPacket(data, data.length, InetAddress.getByAddress(ip), udpPort));
+                }catch (Exception ignored){}
+                ip[3] = -1; //255.255.255.0
+                try{
+                    socket.send(new DatagramPacket(data, data.length, InetAddress.getByAddress(ip), udpPort));
+                }catch (Exception ignored){}
+                ip[2] = -1; //255.255.0.0
+                try{
+                    socket.send(new DatagramPacket(data, data.length, InetAddress.getByAddress(ip), udpPort));
+                }catch (Exception ignored){}
+            }
+        }
+    }
+
+    private boolean isLocal(InetAddress addr) {
+        if (addr.isAnyLocalAddress() || addr.isLoopbackAddress()) return true;
+
+        try {
+            return NetworkInterface.getByInetAddress(addr) != null;
+        } catch (Exception e) {
+            return false;
+        }
+    }
+}

net/src/io/anuke/mindustry/net/MServer.java

@@ -0,0 +1,120 @@
+package io.anuke.mindustry.net;
+
+import io.anuke.arc.*;
+import io.anuke.arc.util.*;
+import io.anuke.mindustry.game.EventType.*;
+import io.anuke.mindustry.net.Net.*;
+import io.anuke.mindustry.net.Packets.*;
+import io.anuke.mnet.*;
+import io.anuke.mnet.MServerSocket;
+
+import java.io.*;
+import java.net.*;
+import java.nio.*;
+import java.util.concurrent.*;
+
+public class MServer implements ServerProvider, ApplicationListener{
+    final CopyOnWriteArrayList<MConnectionImpl> connections = new CopyOnWriteArrayList<>();
+    MServerSocket socket;
+
+    public MServer(){
+        Events.on(AppLoadEvent.class, e -> {
+            Core.app.addListener(this);
+        });
+    }
+
+    public void host(int port) throws IOException{
+        socket = new MServerSocket(port, con -> {
+            MSocket sock = con.accept(null);
+
+            MConnectionImpl kn = new MConnectionImpl(sock);
+            sock.setUserData(kn);
+
+            String ip = sock.getRemoteAddress().getHostAddress();
+
+            Connect c = new Connect();
+            c.id = kn.id;
+            c.addressTCP = ip;
+
+            Log.info("&bRecieved connection: {0} / {1}", c.id, c.addressTCP);
+
+            connections.add(kn);
+            Core.app.post(() -> Net.handleServerReceived(kn.id, c));
+
+            sock.addDcListener((socket, message) -> {
+                Log.info("&bLost connection {0}. Reason: {1}", kn.id, message);
+
+                Disconnect dc = new Disconnect();
+                dc.id = kn.id;
+
+                Core.app.post(() -> {
+                    Net.handleServerReceived(kn.id, dc);
+                    connections.remove(kn);
+                });
+            });
+        }, PacketSerializer::new, () -> {
+            ByteBuffer buf = NetworkIO.writeServerData();
+            byte[] bytes = buf.array();
+            return new DatagramPacket(bytes, bytes.length);
+        });
+
+        connections.clear();
+    }
+
+    public void update(){
+        if(socket == null) return;
+
+        socket.update();
+        for(MSocket socket : socket.getSockets()){
+            MConnectionImpl c = socket.getUserData();
+            socket.update((s, msg) -> Core.app.post(() -> {
+                try{
+                    Net.handleServerReceived(c.id, msg);
+                }catch(Exception e){
+                    e.printStackTrace();
+                }
+            }));
+        }
+    }
+
+    public void close(){
+        if(socket != null) socket.close();
+    }
+
+    public Iterable<? extends NetConnection> getConnections(){
+        return connections;
+    }
+
+    public MConnectionImpl getByID(int id){
+        for(MConnectionImpl n : connections){
+            if(n.id == id){
+                return n;
+            }
+        }
+        return null;
+    }
+
+    class MConnectionImpl extends NetConnection{
+        private final MSocket sock; //sock.
+
+        public MConnectionImpl(MSocket con){
+            super(con.getRemoteAddress().getHostAddress());
+            this.sock = con;
+        }
+
+        @Override
+        public void send(Object object, SendMode mode){
+            if(mode == SendMode.tcp){
+                sock.send(object);
+            }else{
+                sock.sendUnreliable(object);
+            }
+
+        }
+
+        @Override
+        public void close(){
+            sock.close();
+        }
+    }
+}

net/src/io/anuke/mindustry/net/PacketSerializer.java

@@ -1,42 +1,85 @@
 package io.anuke.mindustry.net;
 
 import io.anuke.arc.function.Supplier;
-import io.anuke.arc.net.FrameworkMessage;
+import io.anuke.arc.net.*;
 import io.anuke.arc.net.FrameworkMessage.*;
-import io.anuke.arc.net.NetSerializer;
-import io.anuke.arc.util.pooling.Pools;
+import io.anuke.arc.util.pooling.*;
+import io.anuke.mnet.*;
 
-import java.nio.ByteBuffer;
+import java.nio.*;
+import java.util.*;
 
 @SuppressWarnings("unchecked")
-public class PacketSerializer implements NetSerializer{
+public class PacketSerializer implements NetSerializer, MSerializer{
+    private ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024 * 8);
 
     @Override
     public void write(ByteBuffer byteBuffer, Object o){
-        if(o instanceof FrameworkMessage){
-            byteBuffer.put((byte)-2); //code for framework message
-            writeFramework(byteBuffer, (FrameworkMessage)o);
-        }else{
-            if(!(o instanceof Packet))
-                throw new RuntimeException("All sent objects must implement be Packets! Class: " + o.getClass());
-            byte id = Registrator.getID(o.getClass());
-            if(id == -1)
-                throw new RuntimeException("Unregistered class: " + o.getClass());
-            byteBuffer.put(id);
-            ((Packet)o).write(byteBuffer);
+        if(o == null){
+            byteBuffer.put((byte)-1);
+            return;
         }
+
+        if (!(o instanceof Packet))
+            throw new RuntimeException("All sent objects must implement be Packets! Class: " + o.getClass());
+        byte id = Registrator.getID(o.getClass());
+        if (id == -1)
+            throw new RuntimeException("Unregistered class: " + o.getClass());
+        byteBuffer.put(id);
+        ((Packet) o).write(byteBuffer);
     }
 
     @Override
     public Object read(ByteBuffer byteBuffer){
         byte id = byteBuffer.get();
-        if(id == -2){
-            return readFramework(byteBuffer);
-        }else{
-            Packet packet = Pools.obtain((Class<Packet>)Registrator.getByID(id).type, (Supplier<Packet>)Registrator.getByID(id).constructor);
-            packet.read(byteBuffer);
-            return packet;
+        if(id == -1){
+            return null;
         }
+        Packet packet = Pools.obtain((Class<Packet>) Registrator.getByID(id).type, (Supplier<Packet>) Registrator.getByID(id).constructor);
+        packet.read(byteBuffer);
+        return packet;
+    }
+
+    @Override
+    public byte[] serialize(Object o){
+        buffer.position(0);
+        write(buffer, o);
+        return Arrays.copyOfRange(buffer.array(), 0, buffer.position());
+    }
+
+    @Override
+    public byte[] serialize(Object o, int offset){
+        buffer.position(0);
+        write(buffer, o);
+        int length = buffer.position();
+        byte[] bytes = new byte[length + offset];
+        System.arraycopy(buffer.array(), 0, bytes, offset, length);
+        return bytes;
+    }
+
+    @Override
+    public int serialize(Object o, byte[] bytes, int offset){
+        buffer.position(0);
+        write(buffer, o);
+        int length = buffer.position();
+        System.arraycopy(buffer.array(), 0, bytes, offset, length);
+        return length;
+    }
+
+    @Override
+    public Object deserialize(byte[] bytes){
+        buffer.position(0);
+        buffer.put(bytes);
+        buffer.position(0);
+        return read(buffer);
+    }
+
+    @Override
+    public Object deserialize(byte[] bytes, int offset, int length){
+        buffer.position(0);
+        buffer.put(bytes, offset, length);
+        buffer.position(0);
+        return read(buffer);
     }