diff --git a/lualib/mqueue.lua b/lualib/mqueue.lua index a0913eff..d266b852 100644 --- a/lualib/mqueue.lua +++ b/lualib/mqueue.lua @@ -20,21 +20,8 @@ skynet.register_protocol { end } --- todo: write to log system -local function write_log(...) - return print(...) -end - -local function report_error(succ, ...) - if succ then - return ... - else - write_log("Message queue dispatch error: ", ...) - end -end - local function do_func(f, msg) - return report_error(pcall(f, table.unpack(msg))) + return pcall(f, table.unpack(msg)) end local function message_dispatch(f) @@ -47,7 +34,8 @@ local function message_dispatch(f) local session = msg.session if session == 0 then if do_func(f, msg) ~= nil then - write_log("Message queue returns value, maybe need call this service") + skynet.fork(message_dispatch,f) + error(string.format("[:%x] send a message to [:%x] return something", msg.addr, skynet.self())) end else local data, size = skynet.pack(do_func(f,msg)) @@ -65,7 +53,7 @@ function mqueue.register(f) end function mqueue.call(addr, ...) - return skynet.call(addr, "queue", ...) + return assert(skynet.call(addr, "queue", ...)) end function mqueue.send(addr, ...) diff --git a/lualib/skynet.lua b/lualib/skynet.lua index f3fb4fac..db7906cb 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -5,6 +5,7 @@ local tonumber = tonumber local coroutine = coroutine local assert = assert local pairs = pairs +local pcall = pcall local proto = {} local skynet = {} @@ -292,7 +293,7 @@ function skynet.fork(func,...) table.insert(fork_queue, co) end -local function dispatch_message(prototype, msg, sz, session, source, ...) +local function raw_dispatch_message(prototype, msg, sz, session, source, ...) -- PTYPE_RESPONSE = 1, read skynet.h if prototype == 1 then local co = session_id_coroutine[session] @@ -318,14 +319,27 @@ local function dispatch_message(prototype, msg, sz, session, source, ...) error(string.format("Can't dispatch type %s : ", p.name)) end end +end + +local function dispatch_message(...) + local succ, err = pcall(raw_dispatch_message,...) while true do local key,co = next(fork_queue) if co == nil then - return + break end fork_queue[key] = nil - suspend(co,coroutine.resume(co)) + local fork_succ, fork_err = pcall(suspend,co,coroutine.resume(co)) + if not fork_succ then + if succ then + succ = false + err = fork_err + else + err = err .. "\n" .. fork_err + end + end end + assert(succ, err) end function skynet.newservice(name, ...)