fixed network, added packet split

hxd/net/SteamHost.hx

@@ -29,21 +29,41 @@ package hxd.net;
 #end
 import hxbit.NetworkHost;
 
+@:allow(hxd.net.SteamHost)
 class SteamClient extends NetworkClient {
 
+	static var MAX_PACKET_SIZE = 512 * 1024;
+	static var MAX_PAYLOAD_SIZE = MAX_PACKET_SIZE - 32;
+
 	public var user : steam.User;
 
+	var bigPacket : haxe.io.Bytes;
+	var bigPacketPosition = 0;
+
 	public function new(host, u) {
 		super(host);
 		this.user = u;
 	}
 
-	override function send( bytes : haxe.io.Bytes ) {
-		steam.Networking.sendP2P(user, bytes, Reliable);
+	override function send( data : haxe.io.Bytes ) {
+		if( data.length > MAX_PACKET_SIZE ) {
+			// split
+			var bsize = haxe.io.Bytes.alloc(4);
+			bsize.setInt32(0, data.length);
+			host.sendCustom(SteamHost.CBIG_PACKET, bsize, this);
+			var split = Std.int(data.length / MAX_PAYLOAD_SIZE);
+			for( i in 0...split )
+				host.sendCustom(SteamHost.CBIG_PACKET_DATA, data.sub(i * MAX_PAYLOAD_SIZE, MAX_PAYLOAD_SIZE), this);
+			host.sendCustom(SteamHost.CBIG_PACKET_DATA, data.sub(split * MAX_PAYLOAD_SIZE, data.length - split * MAX_PAYLOAD_SIZE), this);
+			return;
+		}
+		//Sys.println(user + " > [" + data.length+"] " + (data.length > 100 ? data.sub(0,60).toHex()+"..."+data.sub(data.length-8,8).toHex() : data.toHex()));
+		steam.Networking.sendP2P(user, data, Reliable);
 	}
 
 	override function stop() {
 		super.stop();
+		Sys.println("CLOSE " + user);
 		steam.Networking.closeSession(user);
 	}
 
@@ -51,6 +71,11 @@ class SteamClient extends NetworkClient {
 
 class SteamHost extends NetworkHost {
 
+	public static inline var CHELLO_CLIENT = 0;
+	public static inline var CHELLO_SERVER = 1;
+	public static inline var CBIG_PACKET = 2;
+	public static inline var CBIG_PACKET_DATA = 3;
+
 	public var enableSound : Bool = true;
 	var server : steam.User;
 	var onConnected : Bool -> Void;
@@ -73,7 +98,6 @@ class SteamHost extends NetworkHost {
 	}
 
 	public function startClient( server : steam.User, onConnected : Bool -> Void ) {
-		close();
 		isAuth = false;
 		this.server = server;
 		clients = [new SteamClient(this, server)];
@@ -83,14 +107,36 @@ class SteamHost extends NetworkHost {
 	}
 
 	override function onCustom(from:NetworkClient, id:Int, ?data:haxe.io.Bytes) {
-		if( !isAuth && id == 1 && from == clients[0] )
+		switch( id ) {
+		case CHELLO_CLIENT if( isAuth ):
+			sendCustom(CHELLO_SERVER, from);
+		case CHELLO_SERVER if( !isAuth && from == clients[0] ):
 			onConnected(true);
-		if( isAuth && id == 0 )
-			sendCustom(1, from);
+		case CBIG_PACKET:
+			var from = Std.instance(from, SteamClient);
+			from.bigPacket = haxe.io.Bytes.alloc(data.getInt32(0));
+			from.bigPacketPosition = 0;
+		case CBIG_PACKET_DATA:
+			var from = Std.instance(from, SteamClient);
+			from.bigPacket.blit(from.bigPacketPosition, data, 0, data.length);
+			from.bigPacketPosition += data.length;
+			if( from.bigPacketPosition == from.bigPacket.length ) {
+				var data = from.bigPacket;
+				from.bigPacket = null;
+				@:privateAccess {
+					var oldIn = ctx.input;
+					var oldPos = ctx.inPos;
+					onData(from.user, data);
+					ctx.input = oldIn;
+					ctx.inPos = oldPos;
+				}
+			}
+		default:
+			throw "Unknown custom packet " + id;
+		}
 	}
 
 	public function startServer() {
-		close();
 		this.server = steam.Api.getUser();
 		steam.Networking.startP2P(this);
 		isAuth = true;
@@ -105,6 +151,9 @@ class SteamHost extends NetworkHost {
 		for( c in clients )
 			if( Std.instance(c, SteamClient).user == user )
 				return c;
+		for( c in pendingClients )
+			if( Std.instance(c, SteamClient).user == user )
+				return c;
 		return null;
 	}
 
@@ -120,25 +169,14 @@ class SteamHost extends NetworkHost {
 	}
 
 	function onData( from : steam.User, data : haxe.io.Bytes ) {
+		//Sys.println(from + " < [" + data.length+"] " + (data.length > 100 ? data.sub(0,60).toHex()+"..."+data.sub(data.length-8,8).toHex() : data.toHex()));
 		var c = getClient(from);
 		if( c == null ) {
 			c = new SteamClient(this, from);
-			clients.push(c);
-		}
-		// reuse BytesInput (no allocation)
-		@:privateAccess {
-			#if hl
-			input.b = data;
-			input.len = input.totlen = data.length;
-			#else
-			throw "TODO:impl";
-			#end
-		}
-		input.position = 0;
-		while( @:privateAccess c.readData(input, input.length - input.position) ) {
+			pendingClients.push(c);
 		}
+		@:privateAccess c.processMessagesData(data, data.length);
 	}
 
-	// ---
 
 }