-

src/common/haxe/ru/m/armageddon/core/connect/BaseConnection.hx

@@ -1,5 +1,6 @@
 package ru.m.armageddon.core.connect;
 
+import haxe.io.Bytes;
 import protohx.Message;
 import ru.m.armageddon.core.connect.IConnection;
 
@@ -13,5 +14,7 @@ class BaseConnection implements IConnection {
     this.handler = handler;
   }
 
+  public function pushData(bytes:Bytes):Void {}
   public function send(packet:Message):Void {}
+  private function receive(packet:Message):Void {}
 }

src/common/haxe/ru/m/armageddon/core/connect/IConnection.hx

@@ -1,11 +1,14 @@
 package ru.m.armageddon.core.connect;
 
+import haxe.io.Bytes;
 import protohx.Message;
 
 interface IConnection {
   private var builder:IPacketBuilder;
   private var handler:IConnectionHandler;
   public function send(packet:Message):Void;
+  public function pushData(bytes:Bytes):Void;
+  private function receive(packet:Message):Void;
 }
 
 interface IConnectionHandler {

src/common/haxe/ru/m/armageddon/core/connect/NekoConnection.hx

@@ -1,99 +0,0 @@
-package ru.m.armageddon.core.connect;
-
-import sys.net.Socket;
-import haxe.io.BytesOutput;
-import protohx.Message;
-import haxe.io.Bytes;
-
-class NekoConnection implements IConnection {
-
-  private static var MAP:Map<Int, Map<Int, Class<Message>>> = [
-  0x01 => [
-  0x0001 = > LoginRequest,
-  0x0002 = > LoginResponse
-  ]
-  ];
-
-  private var socket:Socket;
-
-  public function new(host, port) {
-    this.onConnect = onConnect;
-    this.addBytes = addBytes;
-    this.onClose = onClose;
-    try {
-      socket.connect(new sys.net.Host(host), port);
-    } catch (e:Dynamic) {
-      trace(e);
-      //onClose();
-      return;
-    }
-    onConnect();
-
-    var buffer = Bytes.alloc(1024);
-    var socks = [socket];
-    var timer = new haxe.Timer(100);
-    timer.run = function() {
-      try {
-        var r:Array<sys.net.Socket>;
-        do {
-          r = sys.net.Socket.select(socks, null, null, 0.001).read;
-          for (s in r) {
-            var size = s.input.readBytes(buffer, 0, buffer.length);
-            onSocketData(buffer.sub(0, size));
-          }
-        } while (r.length > 0);
-      } catch (e:haxe.io.Eof) {
-        timer.stop();
-        //onClose();
-        socket.close();
-      } catch (e:Dynamic) {
-        trace(e);
-        //onClose();
-      }
-    };
-  }
-
-  public dynamic function onConnect():Void {
-
-  }
-
-  private function onSocketData(data):Void {
-    var family = socket.readByte();
-    var id = socket.readByte();
-    var b = new flash.utils.ByteArray();
-    socket.readBytes(b);
-    var bs = Bytes.ofData(cast b);
-    var packet = buildPacket(family, id);
-    packet.mergeFrom(bs);
-    receive(packet);
-  }
-
-
-  private function buildPacket(family:Int, id:Int):Message {
-    var packetClass = MAP[family][id];
-    return Type.createInstance(packetClass, []);
-  }
-
-  public function send(packet:Message):Void {
-    for (family in MAP.keys()) {
-      var subMap = MAP[family];
-      for (id in MAP.keys()) {
-        var packetClass = subMap[id];
-        if (Std.is(packet, packetClass)) {
-          socket.output.writeByte(family);
-          socket.output.writeByte(id);
-          var b = new BytesOutput();
-          packet.writeTo(b);
-          var bytes = b.getBytes();
-          socket.output.writeUInt16(bytes.length);
-          socket.output.writeBytes(bytes, 0, bytes.length);
-          socket.output.flush();
-        }
-      }
-    }
-  }
-
-  public dynamic function receive(packet:Message):Void {
-    L.d("Receive", packet + "");
-  }
-}

src/common/haxe/ru/m/armageddon/core/connect/PacketBuilder.hx

@@ -9,12 +9,12 @@ import ru.m.armageddon.proto.ErrorResponse;
 class PacketBuilder implements IPacketBuilder {
 
   private static var MAP:Map<Int, Map<Int, Class<Message>>> = [
+    0x00 => [
+      0x0001 => ErrorResponse
+    ],
     0x01 => [
       0x0001 => LoginRequest,
       0x0002 => LoginResponse
-    ],
-    0x10 => [
-      0x0001 => ErrorResponse
     ]
   ];
 

src/common/haxe/ru/m/armageddon/core/connect/flash/FlashConnection.hx

@@ -1,4 +1,4 @@
-package ru.m.armageddon.core.connect;
+package ru.m.armageddon.core.connect.flash;
 
 import ru.m.armageddon.core.connect.IConnection.IConnectionHandler;
 import protohx.MessageUtils;
@@ -68,9 +68,10 @@ class FlashConnection extends BaseConnection {
     socket.flush();
   }
 
-  private function receive(packet:Message):Void {
+  override private function receive(packet:Message):Void {
     L.d("Receive", MessageUtils.toJson(packet));
-    var name = Type.getClassName(Type.getClass(packet)).split("::")[1];
+    var name = Type.getClassName(Type.getClass(packet)).split(".").pop();
     L.d("xxx", name);
+    handler.onPacket(packet);
   }
 }

src/common/haxe/ru/m/armageddon/core/connect/neko/NekoConnection.hx

@@ -0,0 +1,60 @@
+package ru.m.armageddon.core.connect.neko;
+
+import haxe.io.BytesOutput;
+import protohx.MessageUtils;
+import protohx.Message;
+import haxe.io.Bytes;
+import sys.net.Socket;
+import ru.m.armageddon.core.connect.IConnection.IConnectionHandler;
+
+class NekoConnection extends BaseConnection {
+
+  public var queue(default, null):PacketQueue;
+
+  private var socket:Socket;
+
+  public function new(socket:Socket, handler:IConnectionHandler) {
+    super(handler);
+    this.socket = socket;
+    socket.setFastSend(true);
+    socket.output.bigEndian = false;
+    socket.input.bigEndian = false;
+    queue = new PacketQueue(builder);
+  }
+
+  override public function pushData(bytes:Bytes):Void {
+    queue.addBytes(bytes);
+    while (queue.hasMsg()) {
+      var packet:Message = queue.popMsg();
+      try {
+        receive(packet);
+      } catch (error:Dynamic) {
+        handler.onError(error);
+      }
+    }
+  }
+
+  override public function send(packet:Message):Void {
+    trace("Send: " + MessageUtils.toJson(packet));
+    try {
+      var meta = builder.packetMeta(packet);
+      var b = new BytesOutput();
+      packet.writeTo(b);
+      var bytes = b.getBytes();
+      socket.output.writeByte(meta.family);
+      socket.output.writeByte(meta.id);
+      socket.output.writeUInt16(bytes.length);
+      socket.output.write(bytes);
+      socket.output.flush();
+    } catch (e:Dynamic) {
+      trace(e);
+    }
+  }
+
+  override private function receive(packet:Message):Void {
+    trace("Receive", MessageUtils.toJson(packet));
+    var name = Type.getClassName(Type.getClass(packet)).split(".").pop();
+    trace("xxx", name);
+    handler.onPacket(packet);
+  }
+}

src/common/haxe/ru/m/armageddon/core/connect/neko/PacketQueue.hx

@@ -0,0 +1,80 @@
+package ru.m.armageddon.core.connect.neko;
+
+import ru.m.armageddon.core.connect.IConnection.IPacketBuilder;
+import protohx.Message;
+import haxe.io.BytesInput;
+import haxe.io.BytesBuffer;
+import haxe.io.Bytes;
+
+class PacketQueue {
+
+  public static inline var HEADER_SIZE:Int = 4;
+
+  private var builder:IPacketBuilder;
+  private var bytesBuff:Bytes;
+  private var msgs:List<Message>;
+
+  public function new(builder:IPacketBuilder) {
+    this.builder = builder;
+    msgs = new List<Message>();
+  }
+
+  public inline function hasMsg():Bool {
+    return !msgs.isEmpty();
+  }
+
+  public inline function popMsg():Message {
+    return msgs.pop();
+  }
+
+  public inline function addMsg(msg:Message):Void {
+    msgs.add(msg);
+  }
+
+  public function addBytes(bytes:Bytes) {
+    if (bytes == null) {
+      return;
+    }
+    if (bytesBuff == null) {
+      bytesBuff = bytes;
+    } else {
+      var buffer = new BytesBuffer();
+      buffer.add(bytesBuff);
+      buffer.add(bytes);
+      bytesBuff = buffer.getBytes();
+    }
+    if (bytesBuff == null || bytesBuff.length < HEADER_SIZE) {
+      return;
+    }
+    var available = bytesBuff.length;
+    var bi = new BytesInput(bytesBuff);
+    bi.bigEndian = false;
+    while (available >= HEADER_SIZE) {
+      var family = bi.readByte();
+      var id = bi.readByte();
+      var packetSize = bi.readUInt16();
+      available -= HEADER_SIZE;
+      if (packetSize <= available) {
+        available -= packetSize;
+        var msgBytes = bi.read(packetSize);
+        var packet = builder.buildPacket({family:family, id:id});
+        packet.mergeFrom(msgBytes);
+        addMsg(packet);
+      } else {
+        available += HEADER_SIZE;
+        break;
+      }
+    }
+    if (available == 0) {
+      bytesBuff = null;
+    } else if (available > 0) {
+      if (bytesBuff.length != available) {
+        var pos = bytesBuff.length - available;
+        bytesBuff = bytesBuff.sub(pos, available);
+      }
+    } else {
+      throw "Wrong available: " + available;
+    }
+  }
+
+}