-

proto/base.proto

@@ -1,3 +1,5 @@
+package ru.m.armageddon.proto;
+
 message User {
   required int32 id = 1;
   required string login = 2;

src/client/haxe/ru/m/armageddon/client/Client.hx

@@ -1,10 +1,11 @@
 package ru.m.armageddon.client;
 
+import flash.Lib;
 import flash.events.MouseEvent;
-import flash.events.Event;
 import ru.m.armageddon.core.connect.FlashConnection;
 import ru.m.armageddon.core.connect.IConnection;
 import haxework.log.TraceLogger;
+import ru.m.armageddon.proto.LoginRequest;
 
 class Client {
 
@@ -22,12 +23,12 @@ class Client {
   public function new() {
     connection = new FlashConnection("localhost", 5000, onConnect);
 
-    flash.Lib.current.stage.addEventListener(MouseEvent.CLICK, function(_) {
+    Lib.current.stage.addEventListener(MouseEvent.CLICK, function(_) {
       onConnect(_);
     });
   }
 
-  public function onConnect(event:Event):Void {
+  public function onConnect(?event:Dynamic):Void {
     var request = new LoginRequest();
     request.login = "shmyga";
     request.password = "xkbp8jh9z2";

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

@@ -0,0 +1,41 @@
+package ru.m.armageddon.core;
+
+import protohx.Message;
+import ru.m.armageddon.proto.LoginRequest;
+import ru.m.armageddon.proto.LoginResponse;
+
+typedef PacketMeta = {
+  var family:Int;
+  var id:Int;
+}
+
+class PacketBuilder {
+
+  private static var MAP:Map<Int, Map<Int, Class<Message>>> = [
+    0x01 => [
+      0x0001 => LoginRequest,
+      0x0002 => LoginResponse
+    ]
+  ];
+
+  public static function packetMeta(packet:Message):PacketMeta {
+    for (family in MAP.keys()) {
+      var subMap = MAP[family];
+      for (id in subMap.keys()) {
+        var packetClass = subMap[id];
+        if (Std.is(packet, packetClass)) {
+          return {
+            family:family,
+            id:id
+          }
+        }
+      }
+    }
+    return null;
+  }
+
+  public static function buildPacket(family:Int, id:Int):Message {
+    var packetClass = MAP[family][id];
+    return Type.createInstance(packetClass, []);
+  }
+}

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

@@ -1,5 +1,6 @@
 package ru.m.armageddon.core.connect;
 
+import protohx.MessageUtils;
 import flash.utils.Endian;
 import haxe.io.BytesOutput;
 import protohx.Message;
@@ -10,6 +11,8 @@ import flash.events.Event;
 import flash.events.SecurityErrorEvent;
 import flash.events.IOErrorEvent;
 import flash.net.Socket;
+import ru.m.armageddon.proto.LoginRequest;
+import ru.m.armageddon.proto.LoginResponse;
 
 class FlashConnection implements IConnection {
 
@@ -22,7 +25,7 @@ class FlashConnection implements IConnection {
 
   private var socket:Socket;
 
-  public function new(host:String, port:Int, onConnect:Event->Void) {
+  public function new(host:String, port:Int, onConnect:?Dynamic->Void) {
     this.onConnect = onConnect;
     socket = new Socket();
     socket.addEventListener(IOErrorEvent.IO_ERROR, onError);
@@ -39,7 +42,7 @@ class FlashConnection implements IConnection {
 
   }
 
-  public dynamic function onConnect(event:Event):Void {
+  public dynamic function onConnect(?event:Dynamic):Void {
 
   }
 
@@ -54,39 +57,25 @@ class FlashConnection implements IConnection {
     var b = new flash.utils.ByteArray();
     socket.readBytes(b);
     var bs = Bytes.ofData(cast b);
-    var packet = buildPacket(family, id);
+    var packet = PacketBuilder.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 {
-    L.d("XXX", "send:" + packet);
+    L.d("Send", MessageUtils.toJson(packet));
-    for (family in MAP.keys()) {
+    var meta = PacketBuilder.packetMeta(packet);
-      var subMap = MAP[family];
+    socket.writeByte(meta.family);
-      for (id in subMap.keys()) {
+    socket.writeByte(meta.id);
-        var packetClass = subMap[id];
+    var out = new BytesOutput();
-        if (Std.is(packet, packetClass)) {
+    packet.writeTo(out);
-        L.d("XXX", family + ":" + id);
+    var bytes = out.getBytes();
-          socket.writeByte(family);
+    socket.writeShort(bytes.length);
-          socket.writeByte(id);
+    socket.writeBytes(cast bytes.getData());
-          var out = new BytesOutput();
+    socket.flush();
-          packet.writeTo(out);
-          var bytes = out.getBytes();
-          socket.writeShort(bytes.length);
-          socket.writeBytes(cast bytes.getData());
-          socket.flush();
-        }
-      }
-    }
   }
 
   public dynamic function receive(packet:Message):Void {
-    L.d("Receive", protohx.MessageUtils.toJson(packet));
+    L.d("Receive", MessageUtils.toJson(packet));
   }
 }

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

@@ -1,13 +1,12 @@
 package ru.m.armageddon.core.connect;
 
-import flash.events.Event;
 import protohx.Message;
 
 interface IConnection {
 
   public function send(packet:Message):Void;
 
-  public dynamic function onConnect(event:Event):Void;
+  public dynamic function onConnect(?event:Dynamic):Void;
 
   public dynamic function receive(packet:Message):Void;
 }

src/server/haxe/ru/m/armageddon/server/Server.hx

@@ -1,134 +1,33 @@
 package ru.m.armageddon.server;
 
-import haxe.io.BytesOutput;
+import ru.m.armageddon.server.session.Session;
-import haxe.io.BytesInput;
-import sys.db.Connection;
-import sys.db.Mysql;
-
 import neko.Lib;
 import haxe.io.Bytes;
 import sys.net.Socket;
 import neko.net.ThreadServer;
 
-/*class NativeSession extends Session {
-
-  public var socket:Socket;
-
-  public function new(socket:Socket) {
-    super();
-    this.socket = socket;
-    socket.setFastSend(true);
-    socket.output.bigEndian = false;
-    socket.input.bigEndian = false;
-  }
-
-  public override function close():Void {
-    socket.close();
-  }
-
-  public override function bakeMsg(msg:protohx.Message):BakedMsg {
-    return new BakedMsg(msg);
-  }
-
-  public override function writeMsgBaked(msg:BakedMsg):Void {
-    writeMsg(msg.msg);
-  }
-
-  public override function writeMsg(msg:protohx.Message):Void {
-    try {
-      var bytes = msgToBytes(msg);
-      socket.output.writeUInt16(bytes.length);
-      socket.output.write(bytes);
-      socket.output.flush();
-    } catch (e:Dynamic) {
-      trace(e);
-    }
-  }
-
-  public static function msgToBytes(msg:protohx.Message):haxe.io.Bytes {
-    var b = new BytesOutput();
-    msg.writeTo(b);
-    return b.getBytes();
-  }
-
-}*/
-
-class Session {
-
-  public var id:Int;
-  public var socket:Socket;
-
-  public function new(socket:Socket) {
-    this.socket = socket;
-    socket.setFastSend(true);
-    socket.output.bigEndian = false;
-    socket.input.bigEndian = false;
-  }
-}
-
 class Server extends ThreadServer<Session, Bytes> {
 
-  //var sr:SessionRegistry;
-
   public function new() {
     super();
-    //sr = new SessionRegistry();
   }
 
   override function clientConnected(s:Socket):Session {
     var session = new Session(s);
-    //sr.sessionConnect(session);
     Lib.println("client: " + session.id + " / " + s.peer());
     return session;
   }
 
   override function clientDisconnected(session:Session) {
     Lib.println("client " + Std.string(session.id) + " disconnected");
-    //sr.sessionDisconnect(session);
   }
 
   override function readClientMessage(session:Session, buf:Bytes, pos:Int, len:Int) {
-//        trace("data " + buf.length + ":" + pos + ":" + len);
     return {msg: buf.sub(pos, len), bytes: len};
   }
 
   override function clientMessage(session:Session, bytes:Bytes) {
-    var available = bytes.length;
+    session.pushData(bytes);
-    var bi = new BytesInput(bytes);
-    bi.bigEndian = false;
-    var family = bi.readByte();
-    var id = bi.readByte();
-    trace("Packet("+family+","+id+")");
-    var packetSize = bi.readUInt16();
-    available -= 2;
-    available -= 2;
-
-    var msgBytes = bi.read(packetSize);
-
-    var request = new LoginRequest();
-    request.mergeFrom(msgBytes);
-
-    trace(request);
-    //sr.sessionData(session, bytes);
-
-    var db = new Db();
-    var userData = db.getUser();
-    var response = new LoginResponse();
-    var user = new User();
-    user.login = userData.login;
-    user.nickname = userData.login;
-    response.user = user;
-
-    var socket = session.socket;
-    var b = new BytesOutput();
-    response.writeTo(b);
-    var bytes = b.getBytes();
-
-    socket.output.writeByte(1);
-    socket.output.writeByte(2);
-    socket.output.writeUInt16(bytes.length);
-    socket.output.write(bytes);
-    socket.output.flush();
   }
 
   public static function main() {
@@ -136,28 +35,3 @@ class Server extends ThreadServer<Session, Bytes> {
     server.run("localhost", 5000);
   }
 }
-
-
-class Db {
-
-  private var db:Connection;
-
-  public function new() {
-    db = Mysql.connect({
-    host : "localhost",
-    port : 3306,
-    user : "shmyga",
-    pass : "xkbp8jh9z2",
-    socket : null,
-    database : "armageddon"
-    });
-  }
-
-  public function getUser() {
-    var rset = db.request("SELECT * FROM Users");
-    var user = rset.next();
-    return user;
-  }
-}
-
-

src/server/haxe/ru/m/armageddon/server/db/Db.hx

@@ -0,0 +1,26 @@
+package ru.m.armageddon.server.db;
+
+import sys.db.Mysql;
+import sys.db.Connection;
+
+class Db {
+
+  private var db:Connection;
+
+  public function new() {
+    db = Mysql.connect({
+    host : "localhost",
+    port : 3306,
+    user : "shmyga",
+    pass : "xkbp8jh9z2",
+    socket : null,
+    database : "armageddon"
+    });
+  }
+
+  public function getUser() {
+    var rset = db.request("SELECT * FROM Users");
+    var user = rset.next();
+    return user;
+  }
+}

src/server/haxe/ru/m/armageddon/server/session/PacketQueue.hx

@@ -0,0 +1,78 @@
+package ru.m.armageddon.server.session;
+
+import ru.m.armageddon.core.PacketBuilder;
+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 bytesBuff:Bytes;
+  private var msgs:List<Message>;
+
+  public function new() {
+    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 = PacketBuilder.buildPacket(family, 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;
+    }
+  }
+
+}

src/server/haxe/ru/m/armageddon/server/session/Session.hx

@@ -0,0 +1,72 @@
+package ru.m.armageddon.server.session;
+
+import ru.m.armageddon.server.db.Db;
+import ru.m.armageddon.proto.User;
+import ru.m.armageddon.proto.LoginResponse;
+import ru.m.armageddon.proto.LoginRequest;
+import ru.m.armageddon.core.PacketBuilder;
+import protohx.MessageUtils;
+import haxe.io.Bytes;
+import protohx.Message;
+import ru.m.armageddon.core.connect.IConnection;
+import haxe.io.BytesOutput;
+import sys.net.Socket;
+
+class Session implements IConnection {
+
+  public var id:Int;
+  public var socket:Socket;
+  public var queue:PacketQueue;
+
+  public function new(socket:Socket) {
+    this.socket = socket;
+    socket.setFastSend(true);
+    socket.output.bigEndian = false;
+    socket.input.bigEndian = false;
+    queue = new PacketQueue();
+  }
+
+  public function pushData(bytes:Bytes):Void {
+    queue.addBytes(bytes);
+    while (queue.hasMsg()) {
+      var packet:Message = queue.popMsg();
+      receive(packet);
+    }
+  }
+
+  public dynamic function onConnect(?event:Dynamic):Void {}
+
+  public function send(packet:Message):Void {
+    trace("Send: " + MessageUtils.toJson(packet));
+    try {
+      var meta = PacketBuilder.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);
+    }
+  }
+
+  public dynamic function receive(packet:Message):Void {
+    trace("Receive: " + MessageUtils.toJson(packet));
+    if (Std.is(packet, LoginRequest)) {
+      var request = cast(packet, LoginRequest);
+
+      var db = new Db();
+      var userData = db.getUser();
+
+      var user = new User();
+      user.login = userData.login;
+      user.nickname = userData.login;
+      var response = new LoginResponse();
+      response.user = user;
+      send(response);
+    }
+  }
+}