diff --git a/Makefile b/Makefile index f6f12675..d7bb53cf 100644 --- a/Makefile +++ b/Makefile @@ -41,7 +41,7 @@ jemalloc : $(MALLOC_STATICLIB) # skynet CSERVICE = snlua logger gate master harbor -LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket memory profile +LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket memory profile multicast SKYNET_SRC = skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c \ skynet_server.c skynet_start.c skynet_timer.c skynet_error.c \ @@ -102,6 +102,9 @@ $(LUA_CLIB_PATH)/memory.so : lualib-src/lua-memory.c | $(LUA_CLIB_PATH) $(LUA_CLIB_PATH)/profile.so : lualib-src/lua-profile.c | $(LUA_CLIB_PATH) $(CC) $(CFLAGS) $(SHARED) $^ -o $@ +$(LUA_CLIB_PATH)/multicast.so : lualib-src/lua-multicast.c | $(LUA_CLIB_PATH) + $(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ + clean : rm -f $(SKYNET_BUILD_PATH)/skynet $(CSERVICE_PATH)/*.so $(LUA_CLIB_PATH)/*.so diff --git a/lualib-src/lua-multicast.c b/lualib-src/lua-multicast.c new file mode 100644 index 00000000..16e469de --- /dev/null +++ b/lualib-src/lua-multicast.c @@ -0,0 +1,104 @@ +#include "skynet.h" + +#include +#include +#include + +struct mc_package { + int reference; + uint32_t size; + void *data; +}; + +/* + lightuserdata + integer size + + return lightuserdata, sizeof(struct mc_package *) + */ +static int +mc_packlocal(lua_State *L) { + void * data = lua_touserdata(L, 1); + size_t size = luaL_checkunsigned(L, 2); + if (size != (uint32_t)size) { + return luaL_error(L, "Size should be 32bit integer"); + } + struct mc_package * pack = skynet_malloc(sizeof(struct mc_package)); + pack->reference = 0; + pack->size = (uint32_t)size; + pack->data = data; + struct mc_package ** ret = skynet_malloc(sizeof(*ret)); + *ret = pack; + lua_pushlightuserdata(L, ret); + lua_pushinteger(L, sizeof(ret)); + return 2; +} + +/* + lightuserdata struct mc_package ** + integer size (must be sizeof(struct mc_package *) + + return package, lightuserdata, size + */ +static int +mc_unpacklocal(lua_State *L) { + struct mc_package ** pack = lua_touserdata(L,1); + int sz = luaL_checkinteger(L,2); + if (sz != sizeof(*pack)) { + return luaL_error(L, "Invalid multicast package size %d", sz); + } + lua_settop(L, 1); + lua_pushlightuserdata(L, (*pack)->data); + lua_pushunsigned(L, (*pack)->size); + return 3; +} + +/* + lightuserdata struct mc_package ** + integer reference + */ +static int +mc_bindrefer(lua_State *L) { + struct mc_package ** pack = lua_touserdata(L,1); + int ref = luaL_checkinteger(L,2); + if ((*pack)->reference != 0) { + return luaL_error(L, "Can't bind a multicast package more than once"); + } + (*pack)->reference = ref; + + return 0; +} + +/* + lightuserdata struct mc_package ** + */ +static int +mc_closelocal(lua_State *L) { + struct mc_package **ptr = lua_touserdata(L,1); + struct mc_package *pack = *ptr; + + int ref = __sync_sub_and_fetch(&pack->reference, 1); + if (ref <= 0) { + skynet_free(pack->data); + skynet_free(pack); + if (ref < 0) { + return luaL_error(L, "Invalid multicast package reference %d", ref); + } + } + + return 0; +} + +int +luaopen_multicast_c(lua_State *L) { + luaL_Reg l[] = { + { "pack", mc_packlocal }, + { "unpack", mc_unpacklocal }, + { "bind", mc_bindrefer }, + { "close", mc_closelocal }, + { NULL, NULL }, + }; + luaL_checkversion(L); + luaL_newlib(L,l); + return 1; +} diff --git a/lualib/multicast.lua b/lualib/multicast.lua new file mode 100644 index 00000000..d303976a --- /dev/null +++ b/lualib/multicast.lua @@ -0,0 +1,66 @@ +local skynet = require "skynet" +local c = require "multicast.c" + +local multicastd +local multicast = {} +local dispatch = {} + +local function default_conf(conf) + conf = conf or {} + conf.pack = conf.pack or skynet.pack + conf.unpack = conf.unpack or skynet.unpack + + return conf +end + +function multicast.newchannel(conf) + assert(multicastd, "Init first") + local channel = skynet.call(multicastd, "lua", "NEW") + dispatch[channel] = default_conf(conf) + return channel +end + +function multicast.publish(channel, ...) + local conf = assert(dispatch[channel]) + skynet.call(multicastd, "lua", "PUB", channel, c.pack(conf.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) +end + +function multicast.unsubscribe(channel) + assert(multicastd, "Init first") + assert(dispatch[channel]) + dispatch[channel] = nil + skynet.call(multicastd, "lua", "USUB", channel) +end + +local function dispatch_subscribe(channel, source, pack, msg, sz) + local conf = dispatch[channel] + if not conf then + c.close(pack) + error ("Unknown channel " .. channel) + end + + local ok, err = pcall(conf.dispatch, channel, source, conf.unpack(msg, sz)) + c.close(pack) + assert(ok, err) +end + +local function init() + multicastd = skynet.uniqueservice "multicastd" + skynet.register_protocol { + name = "multicast", + id = skynet.PTYPE_MULTICAST, + unpack = c.unpack, + dispatch = dispatch_subscribe, + } +end + +skynet.init(init, "multicast") + +return multicast \ No newline at end of file diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 36bb4d6d..6dde1946 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -16,7 +16,7 @@ local skynet = { -- read skynet.h PTYPE_TEXT = 0, PTYPE_RESPONSE = 1, --- PTYPE_MULTICAST = 2, -- DEPRECATED + PTYPE_MULTICAST = 2, PTYPE_CLIENT = 3, PTYPE_SYSTEM = 4, PTYPE_HARBOR = 5, diff --git a/service/multicastd.lua b/service/multicastd.lua new file mode 100644 index 00000000..02b8ef89 --- /dev/null +++ b/service/multicastd.lua @@ -0,0 +1,52 @@ +local skynet = require "skynet" +local mc = require "multicast.c" + +local command = {} +local channel = {} +local channel_n = {} +local channel_id = skynet.harbor(skynet.self()) + +function command.NEW() + channel[channel_id] = {} + channel_n[channel_id] = 0 + local ret = channel_id + channel_id = channel_id + 256 + return ret +end + +function command.PUB(source, c, pack, size) + local group = assert(channel[c]) + mc.bind(pack, channel_n[c]) + local msg = skynet.tostring(pack, size) + for k in pairs(group) do + skynet.redirect(k, source, "multicast", c , msg) + end +end + +function command.SUB(source, c) + local group = assert(channel[c]) + if not group[source] then + channel_n[c] = channel_n[c] + 1 + group[source] = true + end +end + +function command.USUB(source, c) + local group = assert(channel[c]) + if group[source] then + group[source] = nil + channel_n[c] = channel_n[c] - 1 + end +end + +skynet.register_protocol { + name = "multicast", + id = skynet.PTYPE_MULTICAST, +} + +skynet.start(function() + skynet.dispatch("lua", function(_,source, cmd, ...) + local f = assert(command[cmd]) + skynet.ret(skynet.pack(f(source, ...))) + end) +end) diff --git a/test/testmulticast.lua b/test/testmulticast.lua new file mode 100644 index 00000000..55914016 --- /dev/null +++ b/test/testmulticast.lua @@ -0,0 +1,32 @@ +local skynet = require "skynet" +local mc = require "multicast" + +local mode = ... + +if mode == "sub" then + +skynet.start(function() + skynet.dispatch("lua", function (_,_, cmd, channel) + assert(cmd == "init") + mc.subscribe(channel, { + dispatch = function (channel, source, ...) + print(string.format("%s ===> %s (%d)",skynet.address(source), skynet.address(skynet.self()), channel), ...) + end + }) + end) +end) + +else + +skynet.start(function() + local channel = mc.newchannel() + print("New channel", channel) + for i=1,10 do + local sub = skynet.newservice("testmulticast", "sub") + skynet.send(sub, "lua", "init", channel) + end + + mc.publish(channel, "Hello World") +end) + +end \ No newline at end of file