diff --git a/examples/config b/examples/config index 4243a214..a6d9f5ad 100644 --- a/examples/config +++ b/examples/config @@ -7,5 +7,6 @@ master = "127.0.0.1:2013" start = "main" standalone = "0.0.0.0:2013" luaservice = root.."service/?.lua;"..root.."test/?.lua;"..root.."examples/?.lua" +snax = root.."examples/?.lua;"..root.."test/?.lua" cpath = root.."cservice/?.so" redis = root .. "examples/redisconf" diff --git a/lualib/snax.lua b/lualib/snax.lua new file mode 100644 index 00000000..e60aa293 --- /dev/null +++ b/lualib/snax.lua @@ -0,0 +1,109 @@ +local skynet = require "skynet" +local snax_interface = require "snax_interface" + +local snax = {} +local typeclass = {} + +local G = { require = function() end } + +function snax.interface(name) + if typeclass[name] then + return typeclass[name] + end + + local si = snax_interface(name, G) + + local ret = { + subscribe = {}, + response = {}, + system = {}, + } + + for _,v in ipairs(si) do + local id, group, name, f = table.unpack(v) + ret[group][name] = id + end + + typeclass[name] = ret + return ret +end + +local meta = { __tostring = function(v) return string.format("[%s:%x]", v.type, v.handle) end} + +local skynet_send = skynet.send +local skynet_call = skynet.call + +local function gen_pub(type, handle) + return setmetatable({} , { + __index = function( t, k ) + local id = assert(type.subscribe[k] , string.format("publish %s no exist", k)) + return function(...) + skynet_send(handle, "lua", id, ...) + end + end }) +end + +local function gen_req(type, handle) + return setmetatable({} , { + __index = function( t, k ) + local id = assert(type.response[k] , string.format("request %s no exist", k)) + return function(...) + return skynet_call(handle, "lua", id, ...) + end + end }) +end + +local function wrapper(handle, name, type) + return setmetatable ({ + pub = gen_pub(type, handle), + req = gen_req(type, handle), + type = name, + handle = handle, + }, meta) +end + +local handle_cache = setmetatable( {} , { __mode = "kv" } ) + +function snax.newservice(name, ...) + local t = snax.interface(name) + local handle = skynet.newservice("snaxd", name) + assert(handle_cache[handle] == nil) + if t.system.init then + skynet.call(handle, "lua", t.system.init, ...) + end + local ret = wrapper(handle, name, t) + handle_cache[handle] = ret + return ret +end + +function snax.bind(handle, type) + local ret = handle_cache[handle] + if ret then + assert(ret.type == type) + return ret + end + local t = snax.interface(type) + ret = wrapper(handle, type, t) + handle_cache[handle] = ret + return ret +end + +function snax.kill(obj, ...) + local t = snax.interface(obj.type) + skynet_call(obj.handle, "lua", t.system.exit, ...) +end + +local function test_result(ok, ...) + if ok then + return ... + else + error(...) + end +end + +function snax.hotfix(obj, source, ...) + local t = snax.interface(obj.type) + return test_result(skynet_call(obj.handle, "lua", t.system.hotfix, source, ...)) +end + +return snax diff --git a/lualib/snax_hotfix.lua b/lualib/snax_hotfix.lua new file mode 100644 index 00000000..f42ab0c6 --- /dev/null +++ b/lualib/snax_hotfix.lua @@ -0,0 +1,94 @@ +local si = require "snax_interface" +local io = io + +local hotfix = {} + +local function collect_uv(f , uv) + local i = 1 + while true do + local name, value = debug.getupvalue(f, i) + if name == nil then + break + end + local id = debug.upvalueid(f, i) + + if uv[name] then + assert(uv[name].id == id, string.format("ambiguity local value %s", name)) + else + uv[name] = { func = f, index = i, id = id } + + if type(value) == "function" then + collect_uv(value, uv) + end + end + + i = i + 1 + end +end + +local function collect_all_uv(funcs) + local global = {} + for _, v in pairs(funcs) do + if v[4] then + collect_uv(v[4], global) + end + end + + return global +end + +local function loader(source) + return function (filename, ...) + return load(source, "=patch", ...) + end +end + +local function find_func(funcs, group , name) + for _, desc in pairs(funcs) do + local _, g, n = table.unpack(desc) + if group == g and name == n then + return desc + end + end +end + +local dummy_env = {} + +local function patch_func(funcs, global, group, name, f) + local desc = assert(find_func(funcs, group, name) , string.format("Patch mismatch %s.%s", group, name)) + local i = 1 + while true do + local name, value = debug.getupvalue(f, i) + if name == nil then + break + elseif value == nil or value == dummy_env then + local old_uv = global[name] + if old_uv then + debug.upvaluejoin(f, i, old_uv.func, old_uv.index) + end + end + i = i + 1 + end + desc[4] = f +end + +local function inject(funcs, source, ...) + local patch = si("patch", dummy_env, loader(source)) + local global = collect_all_uv(funcs) + + for _, v in pairs(patch) do + local _, group, name, f = table.unpack(v) + if f then + patch_func(funcs, global, group, name, f) + end + end + + local hf = find_func(patch, "system", "hotfix") + if hf and hf[4] then + return hf[4](...) + end +end + +return function (funcs, source, ...) + return pcall(inject, funcs, source, ...) +end diff --git a/lualib/snax_interface.lua b/lualib/snax_interface.lua new file mode 100644 index 00000000..2dbb06b2 --- /dev/null +++ b/lualib/snax_interface.lua @@ -0,0 +1,85 @@ +local skynet = require "skynet" + +return function (name , G, loader) + loader = loader or loadfile + local mainfunc + + local function func_id(id, group) + local tmp = {} + local function count( _, name, func) + if type(name) ~= "string" then + error (string.format("%s method only support string", group)) + end + if type(func) ~= "function" then + error (string.format("%s.%s must be function"), group, name) + end + if tmp[name] then + error (string.format("%s.%s duplicate definition", group, name)) + end + tmp[name] = true + table.insert(id, { #id + 1, group, name, func} ) + end + return setmetatable({}, { __newindex = count }) + end + + do + assert(getmetatable(G) == nil) + assert(G.init == nil) + assert(G.exit == nil) + + assert(G.subscribe == nil) + assert(G.response == nil) + end + + local env = {} + local func = {} + + local system = { "init", "exit", "hotfix" } + + do + for k, v in ipairs(system) do + system[v] = k + func[k] = { k , "system", v } + end + end + + env.subscribe = func_id(func, "subscribe") + env.response = func_id(func, "response") + + local function init_system(t, name, f) + local index = assert(system[name] , string.format("Not support global var %s", name)) + if type(f) ~= "function" then + error (string.format("%s must be a function", name)) + end + func[index][4] = f + end + + setmetatable(G, { __index = env , __newindex = init_system }) + + do + local path = skynet.getenv "snax" + + local errlist = {} + + for pat in string.gmatch(path,"[^;]+") do + local filename = string.gsub(pat, "?", name) + local f , err = loader(filename, "bt", G) + if f then + mainfunc = f + break + else + table.insert(errlist, err) + end + end + + if mainfunc == nil then + error(table.concat(errlist, "\n")) + end + end + + mainfunc() + + setmetatable(G, nil) + + return func +end diff --git a/service/snaxd.lua b/service/snaxd.lua new file mode 100644 index 00000000..59fe756a --- /dev/null +++ b/service/snaxd.lua @@ -0,0 +1,105 @@ +local skynet = require "skynet" +local c = require "skynet.c" +local snax_interface = require "snax_interface" + +local func = snax_interface(tostring(...), _ENV) +local mode +local thread_id +local message_queue = {} +local init = false + +local function do_func(f, msg) + return pcall(f, table.unpack(msg)) +end + +local function dispatch(f, ...) + return skynet.pack(f(...)) +end + +local function message_dispatch() + while true do + if #message_queue==0 then + thread_id = coroutine.running() + skynet.wait() + else + local msg = table.remove(message_queue,1) + local method = msg.method + local f = method[4] + if f then + if method[2] == "subscribe" then + -- no return + local ok, data = pcall(f, table.unpack(msg)) + if not ok then + print(string.format("Error on [:%x] to [:%x] %s", msg.source, skynet.self(), tostring(data))) + end + else + local ok, data, size = pcall(dispatch, f, table.unpack(msg)) + if ok then + -- skynet.PTYPE_RESPONSE == 1 + c.send(msg.source, 1, msg.session, data, size) + if method[2] == "system" then + init = false + skynet.exit() + break + end + else + -- Can't throw error, so print it directly + print(string.format("Error on [:%x] to [:%x] %s", msg.source, skynet.self(), tostring(data))) + c.send(msg.source, skynet.PTYPE_ERROR, msg.session, "") + end + end + end + end + end +end + +local function queue( session, source, method, ...) + table.insert(message_queue, {session = session, source = source, method = method, ... }) + if thread_id then + skynet.wakeup(thread_id) + thread_id = nil + end +end + +skynet.start(function() + skynet.dispatch("lua", function ( session , source , id, ...) + local method = func[id] + + if method[2] == "system" then + local command = method[3] + if command == "hotfix" then + local hotfix = require "snax_hotfix" + skynet.ret(skynet.pack(hotfix(func, ...))) + elseif command == "init" then + assert(not init, "Already init") + local initfunc = method[4] or function() end + mode = initfunc(...) + if mode == "queue" then + skynet.fork(message_dispatch) + end + skynet.ret() + init = true + elseif mode == "queue" then + queue( session, source, method , ...) + else + assert(init, "Never init") + assert(command == "exit") + local exitfunc = method[4] or function() end + exitfunc(...) + skynet.ret() + init = false + skynet.exit() + end + else + assert(init, "Init first") + if mode == "queue" then + queue(session, source, method , ...) + elseif method[2] == "subscribe" then + -- no return + method[4](...) + else + skynet.ret(skynet.pack(method[4](...))) + end + end + end) +end) diff --git a/test/pingserver.lua b/test/pingserver.lua index 49496608..f36166bb 100644 --- a/test/pingserver.lua +++ b/test/pingserver.lua @@ -1,27 +1,29 @@ local skynet = require "skynet" -local command = {} +local i = 0 +local hello = "hello" -function command.PING(hello) +function response.ping(hello) + skynet.sleep(100) return hello end -function command.HELLO() - skynet.sleep(100) - return "hello" +function subscribe.hello() + i = i + 1 + print (i, hello) end -function command.EXIT() - skynet.exit() -end - -function command.ERROR() +function response.error() error "throw an error" end +function init( ... ) + print ("ping server start:", ...) -skynet.start(function() - skynet.dispatch("lua", function(session,addr, cmd, ...) - skynet.ret(skynet.pack(command[cmd](...))) - end) -end) +-- You can return "queue" for queue service mode +-- return "queue" +end + +function exit(...) + print ("ping server exit:", ...) +end diff --git a/test/testping.lua b/test/testping.lua index 59a2ad5d..1244551e 100644 --- a/test/testping.lua +++ b/test/testping.lua @@ -1,10 +1,29 @@ local skynet = require "skynet" +local snax = require "snax" skynet.start(function() - local ps = skynet.uniqueservice("pingserver") - skynet.watch(ps) - print(pcall(skynet.call,ps,"lua","ERROR")) - print(skynet.call(ps, "lua", "PING", "hello")) - print(skynet.call(ps, "lua", "PING", "hay")) - skynet.call(ps, "lua", "EXIT") + local ps = snax.newservice ("pingserver", "hello world") + print(ps.req.ping("foobar")) + print(ps.pub.hello()) + print(pcall(ps.req.error)) + print("Hotfix (i) :", snax.hotfix(ps, [[ + +local i +local hello + +function subscribe.hello() + i = i + 1 + print ("fix", i, hello) +end + +function hotfix(...) + local temp = i + i = 100 + return temp +end + + ]])) + print(ps.pub.hello()) + print(snax.kill(ps,"exit")) + skynet.exit() end)