use fork to dispatch more socket message

lualib-src/lua-netpack.c

@@ -240,19 +240,12 @@ filter_data_(lua_State *L, int fd, uint8_t * buffer, int size) {
 		buffer += need;
 		size -= need;
 		if (size == 0) {
-			if (q == NULL || q->head == q->tail ) {
+			lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
-				lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
+			lua_pushinteger(L, fd);
-				lua_pushinteger(L, fd);
+			lua_pushlightuserdata(L, uc->pack.buffer);
-				lua_pushlightuserdata(L, uc->pack.buffer);
+			lua_pushinteger(L, uc->pack.size);
-				lua_pushinteger(L, uc->pack.size);
+			skynet_free(uc);
-				skynet_free(uc);
+			return 5;
-				return 5;
-			}
-			else{
-				push_data(L, fd, uc->pack.buffer, uc->pack.size, 0);
-				skynet_free(uc);
-				return 1;
-			}
 		}
 		// more data
 		push_data(L, fd, uc->pack.buffer, uc->pack.size, 0);
@@ -281,21 +274,13 @@ filter_data_(lua_State *L, int fd, uint8_t * buffer, int size) {
 		}
 		if (size == pack_size) {
 			// just one package
-			if ( q == NULL || q->head == q->tail) {
+			lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
-				lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
+			lua_pushinteger(L, fd);
-				lua_pushinteger(L, fd);
+			void * result = skynet_malloc(pack_size);
-				void * result = skynet_malloc(pack_size);
+			memcpy(result, buffer, size);
-				memcpy(result, buffer, size);
+			lua_pushlightuserdata(L, result);
-				lua_pushlightuserdata(L, result);
+			lua_pushinteger(L, size);
-				lua_pushinteger(L, size);
+			return 5;
-				return 5;
-			}
-			else{
-				push_data(L, fd, buffer, pack_size, 1);
-				buffer += pack_size;
-				size -= pack_size;
-				return 1;
-			}
 		}
 		// more data
 		push_data(L, fd, buffer, pack_size, 1);

lualib/snax/gateserver.lua

@@ -53,20 +53,32 @@ function gateserver.start(handler)
 
 	local MSG = {}
 
-	function MSG.data(fd, msg, sz)
+	local function dispatch_msg(fd, msg, sz)
 		if connection[fd] then
 			handler.message(fd, msg, sz)
+		else
+			skynet.error(string.format("Drop message from fd (%d) : %s", fd, netpack.tostring(msg,sz)))
 		end
 	end
 
-	function MSG.more()
+	MSG.data = dispatch_msg
-		for fd, msg, sz in netpack.pop, queue do
+
-			if connection[fd] then
+	local function dispatch_queue()
-				handler.message(fd, msg, sz)
+		local fd, msg, sz = netpack.pop(queue)
+		if fd then
+			-- may dispatch even the handler.message blocked
+			-- If the handler.message never block, the queue should be empty, so only fork once and then exit.
+			skynet.fork(dispatch_queue)
+			dispatch_msg(fd, msg, sz)
+
+			for fd, msg, sz in netpack.pop, queue do
+				dispatch_msg(fd, msg, sz)
 			end
 		end
 	end
 
+	MSG.more = dispatch_queue
+
 	function MSG.open(fd, msg)
 		if client_number >= maxclient then
 			socketdriver.close(fd)
@@ -128,4 +140,4 @@ function gateserver.start(handler)
 	end)
 end
 
-return gateserver
+return gateserver