change multicast api

lualib-src/lua-multicast.c

@@ -139,6 +139,15 @@ mc_remote(lua_State *L) {
 	return 2;
 }
 
+static int
+mc_nextid(lua_State *L) {
+	uint32_t id = luaL_checkunsigned(L, 1);
+	id += 256;
+	lua_pushunsigned(L, id);
+
+	return 1;
+}
+
 int
 luaopen_multicast_c(lua_State *L) {
 	luaL_Reg l[] = {
@@ -149,6 +158,7 @@ luaopen_multicast_c(lua_State *L) {
 		{ "remote", mc_remote },
 		{ "packstring", mc_packstring },
 		{ "packremote", mc_packremote },
+		{ "nextid", mc_nextid },
 		{ NULL, NULL },
 	};
 	luaL_checkversion(L);

lualib/multicast.lua

@@ -1,10 +1,24 @@
 local skynet = require "skynet"
-local c = require "multicast.c"
+local mc = require "multicast.c"
 
 local multicastd
 local multicast = {}
 local dispatch = {}
 
+local chan = {}
+local chan_meta = {
+	__index = chan,
+	__gc = function(self)
+		self:unsubscribe()
+		local c = self.channel
+		self.channel = nil
+		dispatch[c] = nil
+	end,
+	__tostring = function (self)
+		return string.format("[Multicast:%x]",self.channel)
+	end,
+}
+
 local function default_conf(conf)
 	conf = conf or {}
 	conf.pack = conf.pack or skynet.pack
@@ -13,48 +27,69 @@ local function default_conf(conf)
 	return conf
 end
 
-function multicast.newchannel(conf)
+function multicast.new(conf)
 	assert(multicastd, "Init first")
-	local channel = skynet.call(multicastd, "lua", "NEW")
-	dispatch[channel] = default_conf(conf)
-	return channel
+	local self = {}
+	conf = conf or self
+	self.channel = conf.channel
+	if self.channel == nil then
+		self.channel = skynet.call(multicastd, "lua", "NEW")
+	end
+	self.__pack = conf.pack or skynet.pack
+	self.__unpack = conf.unpack or skynet.unpack
+	self.__dispatch = conf.dispatch
+
+	return setmetatable(self, chan_meta)
 end
 
-function multicast.bind(channel, conf)
-	assert(multicastd, "Init first")
-	assert(not dispatch[channel])
-	dispatch[channel] = default_conf(conf)
+function chan:delete()
+	local c = assert(self.channel)
+	skynet.send(multicastd, "lua", "DEL", c)
+	self.channel = nil
+	self.__subscribe = nil
 end
 
-function multicast.publish(channel, ...)
-	local conf = assert(dispatch[channel])
-	skynet.call(multicastd, "lua", "PUB", channel, c.pack(conf.pack(...)))
+function chan:publish(...)
+	local c = assert(self.channel)
+	skynet.call(multicastd, "lua", "PUB", c, mc.pack(self.__pack(...)))
 end
 
-function multicast.subscribe(channel, conf)
-	assert(multicastd, "Init first")
-	assert(conf.dispatch)
-	skynet.call(multicastd, "lua", "SUB", channel)
-	dispatch[channel] = default_conf(conf)
+function chan:subscribe()
+	local c = assert(self.channel)
+	if self.__subscribe then
+		-- already subscribe
+		return
+	end
+	skynet.call(multicastd, "lua", "SUB", c)
+	self.__subscribe = true
+	dispatch[c] = self
 end
 
-function multicast.unsubscribe(channel)
-	assert(multicastd, "Init first")
-	assert(dispatch[channel])
-	dispatch[channel] = nil
-	skynet.call(multicastd, "lua", "USUB", channel)
+function chan:unsubscribe()
+	if not self.__subscribe then
+		-- already unsubscribe
+		return
+	end
+	local c = assert(self.channel)
+	skynet.send(multicastd, "lua", "USUB", c)
+	self.__subscribe = nil
 end
 
 local function dispatch_subscribe(channel, source, pack, msg, sz)
-	local conf = dispatch[channel]
-	if not conf then
-		c.close(pack)
+	local self = dispatch[channel]
+	if not self then
+		mc.close(pack)
 		error ("Unknown channel " .. channel)
 	end
 
-	local ok, err = pcall(conf.dispatch, channel, source, conf.unpack(msg, sz))
-	c.close(pack)
+	if self.__subscribe then
+		local ok, err = pcall(self.__dispatch, self, source, self.__unpack(msg, sz))
+		mc.close(pack)
 		assert(ok, err)
+	else
+		-- maybe unsubscribe first, but the message is send out. drop the message unneed
+		mc.close(pack)
+	end
 end
 
 local function init()
@@ -62,7 +97,7 @@ local function init()
 	skynet.register_protocol {
 		name = "multicast",
 		id = skynet.PTYPE_MULTICAST,
-		unpack = c.unpack,
+		unpack = mc.unpack,
 		dispatch = dispatch_subscribe,
 	}
 end

service/multicastd.lua

@@ -9,6 +9,7 @@ local channel = {}
 local channel_n = {}
 local channel_remote = {}
 local channel_id = harbor_id
+local NORET = {}
 
 local function get_address(t, id)
 	local v = assert(datacenter.get("multicast", id))
@@ -18,20 +19,58 @@ end
 
 local node_address = setmetatable({}, { __index = get_address })
 
+-- new LOCAL channel , The low 8bit is the same with harbor_id
 function command.NEW()
+	while channel[channel_id] do
+		channel_id = mc.nextid(channel_id)
+	end
 	channel[channel_id] = {}
 	channel_n[channel_id] = 0
 	local ret = channel_id
-	channel_id = channel_id + 256
+	channel_id = mc.nextid(channel_id)
 	return ret
 end
 
+-- MUST call by the owner node of channel, delete a remote channel
+function command.DELR(source, c)
+	channel[c] = nil
+	channel_n[c] = nil
+	return NORET
+end
+
+-- delete a channel, if the channel is remote, forward the command to the owner node
+-- otherwise, delete the channel, and call all the remote node, DELR
+function command.DEL(source, c)
+	local node = c % 256
+	if node ~= harbor_id then
+		skynet.send(node_address[node], "lua", "DEL", c)
+		return NORET
+	end
+	local remote = channel_remote[c]
+	channel[c] = nil
+	channel_n[c] = nil
+	channel_remote[c] = nil
+	for node in pairs(remote) do
+		skynet.send(node_address[node], "lua", "DELR", c)
+	end
+	return NORET
+end
+
+-- forward multicast message to a node (channel id use the session field)
 local function remote_publish(node, channel, source, ...)
 	skynet.redirect(node_address[node], source, "multicast", channel, ...)
 end
 
+-- publish a message, for local node, use the message pointer (call mc.bind to add the reference)
+-- for remote node, call remote_publish. (call mc.unpack and skynet.tostring to convert message pointer to string)
 local function publish(c , source, pack, size)
-	local group = assert(channel[c])
+	local group = channel[c]
+	if group == nil then
+		-- dead channel, delete the pack
+		mc.bind(pack, 1)
+		mc.close(pack)
+		return
+	end
 	mc.bind(pack, channel_n[c])
 	local msg = skynet.tostring(pack, size)
 	for k in pairs(group) do
@@ -56,6 +95,8 @@ skynet.register_protocol {
 	dispatch = publish,
 }
 
+-- publish a message, if the caller is remote, forward the message to the owner node (by remote_publish)
+-- If the caller is local, call publish
 function command.PUB(source, c, pack, size)
 	assert(skynet.harbor(source) == harbor_id)
 	local node = c % 256
@@ -67,9 +108,17 @@ function command.PUB(source, c, pack, size)
 	end
 end
 
+-- the node (source) subscribe a channel
+-- MUST call by channel owner node (assert source is not local and channel is create by self)
+-- If channel is not exist, return true
+-- Else set channel_remote[channel] true
 function command.SUBR(source, c)
 	local node = skynet.harbor(source)
-	assert(node ~= harbor_id)
+	if not channel[c] then
+		-- channel none exist
+		return true
+	end
+	assert(node ~= harbor_id and c % 256 == harbor_id)
 	local group = channel_remote[c]
 	if group == nil then
 		group = {}
@@ -78,30 +127,37 @@ function command.SUBR(source, c)
 	group[node] = true
 end
 
+-- the service (source) subscribe a channel
+-- If the channel is remote, node subscribe it by send a SUBR to the owner .
 function command.SUB(source, c)
 	local node = c % 256
 	if node ~= harbor_id then
 		-- remote group
 		if channel[c] == nil then
+			if skynet.call(node_address[node], "lua", "SUBR", c) then
+				return
+			end
 			channel[c] = {}
 			channel_n[c] = 0
-			skynet.call(node_address[node], "lua", "SUBR", c)
 		end
 	end
-	local group = assert(channel[c])
-	if not group[source] then
+	local group = channel[c]
+	if group and not group[source] then
 		channel_n[c] = channel_n[c] + 1
 		group[source] = true
 	end
 end
 
+-- MUST call by a node, unsubscribe a channel
 function command.USUBR(source, c)
 	local node = skynet.harbor(source)
 	assert(node ~= harbor_id)
 	local group = assert(channel_remote[c])
 	group[node] = nil
+	return NORET
 end
 
+-- Unsubscribe a channel, if the subscriber is empty and the channel is remote, send USUBR to the channel owner
 function command.USUB(source, c)
 	local group = assert(channel[c])
 	if group[source] then
@@ -113,16 +169,20 @@ function command.USUB(source, c)
 				-- remote group
 				channel[c] = nil
 				channel_n[c] = nil
-				skynet.call(node_address[node], "lua", "USUBR", c)
+				skynet.send(node_address[node], "lua", "USUBR", c)
 			end
 		end
 	end
+	return NORET
 end
 
 skynet.start(function()
 	skynet.dispatch("lua", function(_,source, cmd, ...)
 		local f = assert(command[cmd])
-		skynet.ret(skynet.pack(f(source, ...)))
+		local result = f(source, ...)
+		if result ~= NORET then
+			skynet.ret(skynet.pack(result))
+		end
 	end)
 	local self = skynet.self()
 	local id = skynet.harbor(self)

test/testmulticast.lua

@@ -9,11 +9,14 @@ if mode == "sub" then
 skynet.start(function()
 	skynet.dispatch("lua", function (_,_, cmd, channel)
 		assert(cmd == "init")
-		mc.subscribe(channel, {
+		local c = mc.new {
+			channel = channel ,
 			dispatch = function (channel, source, ...)
-				print(string.format("%s <=== %s (%d)",skynet.address(skynet.self()),skynet.address(source), channel), ...)
+				print(string.format("%s <=== %s %s",skynet.address(skynet.self()),skynet.address(source), channel), ...)
 			end
-		})
+		}
+		print(skynet.address(skynet.self()), "sub", c)
+		c:subscribe()
 		skynet.ret(skynet.pack())
 	end)
 end)
@@ -21,18 +24,17 @@ end)
 else
 
 skynet.start(function()
-	local channel = mc.newchannel()
+	local channel = mc.new()
 	print("New channel", channel)
 	for i=1,10 do
 		local sub = skynet.newservice("testmulticast", "sub")
-		skynet.call(sub, "lua", "init", channel)
+		skynet.call(sub, "lua", "init", channel.channel)
 	end
 
-	print("set channel", channel)
-	dc.set("CHANNEL", channel)
+	dc.set("MCCHANNEL", channel.channel)	-- for multi node test
 
 	print(skynet.address(skynet.self()), "===>", channel)
-	mc.publish(channel, "Hello World")
+	channel:publish("Hello World")
 end)
 
 end