Merge pull request #89 from cloudwu/snax

snax :  : a simple skynet auxliliary framework

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"

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

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

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

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)

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

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)