add cluster.proxy and cluster.ncall

HISTORY.md

@@ -7,6 +7,8 @@ Dev version
 * Don't check imported function in snax.hotfix
 * snax service add change SERVICE_PATH and add it to package.path
 * skynet.redirect support string address
+* add skynet.harbor.queryname to query globalname
+* add cluster.proxy and cluster.ncall
 
 v0.5.2 (2014-8-11)
 -----------

examples/cluster2.lua

@@ -2,5 +2,8 @@ local skynet = require "skynet"
 local cluster = require "cluster"
 
 skynet.start(function()
-	print(cluster.call("db", "SIMPLEDB", "GET", "a"))
+	local proxy = cluster.proxy("db", ".simpledb")
+	print(skynet.call(proxy, "lua", "GET", "a"))
+	print(cluster.ncall("db.simpledb","GET", "a"))
+	print(cluster.call("db", ".simpledb", "GET", "a"))
 end)

examples/globallog.lua

@@ -4,5 +4,6 @@ skynet.start(function()
 	skynet.dispatch("lua", function(session, address, ...)
 		print("[GLOBALLOG]", skynet.address(address), ...)
 	end)
+	skynet.register ".log"
 	skynet.register "LOG"
 end)

examples/simpledb.lua

@@ -22,5 +22,6 @@ skynet.start(function()
 			error(string.format("Unknown command %s", tostring(cmd)))
 		end
 	end)
+	skynet.register ".simpledb"
 	skynet.register "SIMPLEDB"
 end)

lualib-src/lua-skynet.c

@@ -73,10 +73,17 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t
 	return 0;
 }
 
+static int
+forward_cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
+	_cb(context, ud, type, session, source, msg, sz);
+	// don't delete msg in forward mode.
+	return 1;
+}
+
 static int
 _callback(lua_State *L) {
 	struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
-
+	int forward = lua_toboolean(L, 2);
 	luaL_checktype(L,1,LUA_TFUNCTION);
 	lua_settop(L,1);
 	lua_rawsetp(L, LUA_REGISTRYINDEX, _cb);
@@ -84,7 +91,11 @@ _callback(lua_State *L) {
 	lua_rawgeti(L, LUA_REGISTRYINDEX, LUA_RIDX_MAINTHREAD);
 	lua_State *gL = lua_tothread(L,-1);
 
-	skynet_callback(context, gL, _cb);
+	if (forward) {
+		skynet_callback(context, gL, forward_cb);
+	} else {
+		skynet_callback(context, gL, _cb);
+	}
 
 	return 0;
 }

lualib/cluster.lua

@@ -4,7 +4,7 @@ local clusterd
 local cluster = {}
 
 function cluster.call(node, address, ...)
-	-- skynet.pack(...) will free by cluster.c.packrequest
+	-- skynet.pack(...) will free by cluster.core.packrequest
 	return skynet.call(clusterd, "lua", "req", node, address, skynet.pack(...))
 end
 
@@ -20,6 +20,23 @@ function cluster.reload()
 	skynet.call(clusterd, "lua", "reload")
 end
 
+function cluster.proxy(node, name)
+	return skynet.call(clusterd, "lua", "proxy", node, name)
+end
+
+local namecache = {}
+
+function cluster.ncall(name, ...)
+	local s = namecache[name]
+	if not s then
+		local node , lname = name:match "(.-)(%..+)"
+		assert(node and lname)
+		s = cluster.proxy(node, lname)
+		namecache[name] = assert(s)
+	end
+	return skynet.call(s, "lua", ...)
+end
+
 skynet.init(function()
 	clusterd = skynet.uniqueservice("clusterd")
 end)

lualib/skynet.lua

@@ -99,6 +99,7 @@ local coroutine_yield = coroutine.yield
 local function co_create(f)
 	local co = table.remove(coroutine_pool)
 	if co == nil then
+		local print = print
 		co = coroutine.create(function(...)
 			f(...)
 			while true do
@@ -548,7 +549,7 @@ end
 
 function skynet.address(addr)
 	if type(addr) == "number" then
-		return string.format(":%x",addr)
+		return string.format(":%08x",addr)
 	else
 		return tostring(addr)
 	end
@@ -648,6 +649,21 @@ function skynet.filter(f ,start_func)
 	end)
 end
 
+function skynet.forward_type(map, start_func)
+	c.callback(function(ptype, msg, sz, ...)
+		local prototype = map[ptype]
+		if prototype then
+			dispatch_message(prototype, msg, sz, ...)
+		else
+			dispatch_message(ptype, msg, sz, ...)
+			c.trash(msg, sz)
+		end
+	end, true)
+	skynet.timeout(0, function()
+		init_service(start_func)
+	end)
+end
+
 function skynet.endless()
 	return c.command("ENDLESS")~=nil
 end

service/cdummy.lua

@@ -36,7 +36,11 @@ function harbor.REGISTER(name, handle)
 	skynet.redirect(harbor_service, handle, "harbor", 0, "N " .. name)
 end
 
-function harbor.QUERYNAME(fd, name)
+function harbor.QUERYNAME(name)
+	if name:byte() == 46 then	-- "." , local name
+		skynet.ret(skynet.pack(skynet.localname(name)))
+		return
+	end
 	local result = globalname[name]
 	if result then
 		skynet.ret(skynet.pack(result))

service/clusterd.lua

@@ -60,6 +60,16 @@ function command.req(source, node, addr, msg, sz)
 	skynet.ret(c:request(request, session))
 end
 
+local proxy = {}
+
+function command.proxy(source, node, name)
+	local fullname = node .. "." .. name
+	if proxy[fullname] == nil then
+		proxy[fullname] = skynet.newservice("clusterproxy", node, name)
+	end
+	skynet.ret(skynet.pack(proxy[fullname]))
+end
+
 local request_fd = {}
 
 function command.socket(source, subcmd, fd, msg)

service/cslave.lua

@@ -93,6 +93,7 @@ local function monitor_master(master_fd)
 				local fd = slaves[id_name]
 				slaves[id_name] = false
 				if fd then
+					monitor_clear(id_name)
 					socket.close(fd)
 				end
 			end
@@ -143,14 +144,14 @@ local function monitor_harbor(master_fd)
 	return function(session, source, command)
 		local t = string.sub(command, 1, 1)
 		local arg = string.sub(command, 3)
-		if t == "Q" then
+		if t == 'Q' then
 			-- query name
 			if globalname[arg] then
 				skynet.redirect(harbor_service, globalname[arg], "harbor", 0, "N " .. arg)
 			else
 				socket.write(master_fd, pack_package("Q", arg))
 			end
-		elseif t == "D" then
+		elseif t == 'D' then
 			-- harbor down
 			local id = tonumber(arg)
 			if slaves[id] then
@@ -194,6 +195,10 @@ function harbor.CONNECT(fd, id)
 end
 
 function harbor.QUERYNAME(fd, name)
+	if name:byte() == 46 then	-- "." , local name
+		skynet.ret(skynet.pack(skynet.localname(name)))
+		return
+	end
 	local result = globalname[name]
 	if result then
 		skynet.ret(skynet.pack(result))

skynet-src/skynet_server.c

@@ -232,7 +232,7 @@ _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
 	size_t sz = msg->sz & HANDLE_MASK;
 	if (!ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz)) {
 		skynet_free(msg->data);
-	}
+	} 
 	CHECKCALLING_END(ctx)
 }