rewrite watchdog/agent, use new gate

examples/agent.lua

@@ -1,5 +1,7 @@
 local skynet = require "skynet"
-local client = ...
+local client
+
+local CMD = {}
 
 skynet.register_protocol {
 	name = "client",
@@ -8,12 +10,25 @@ skynet.register_protocol {
 	unpack = skynet.tostring,
 	dispatch = function (session, address, text)
 		-- It's client, there is no session
-		--	skynet.send("LOG", "text", "client message :" .. text)
 		local result = skynet.call("SIMPLEDB", "text", text)
 		skynet.ret(result)
 	end
 }
 
-skynet.start(function()
+function CMD.start(gate , fd)
+	client = skynet.launch("client", fd)
+	skynet.call(gate, "lua", "forward", fd, client)
 	skynet.send(client,"text","Welcome to skynet")
+end
+
+function CMD.exit()
+	skynet.kill(client)
+	client = nil
+end
+
+skynet.start(function()
+	skynet.dispatch("lua", function(_,_, command, ...)
+		local f = CMD[command]
+		skynet.ret(skynet.pack(f(...)))
+	end)
 end)

examples/main.lua

@@ -9,9 +9,12 @@ skynet.start(function()
 	local lualog = skynet.newservice("lualog")
 	local console = skynet.newservice("console")
 --	skynet.newservice("debug_console",8000)
-	local watchdog = skynet.newservice("watchdog","8888", max_client, 0)
-	local db = skynet.newservice("simpledb")
---	skynet.launch("snlua","testgroup")
+	skynet.newservice("simpledb")
+	local watchdog = skynet.newservice("watchdog")
+	skynet.call(watchdog, "lua", "start", {
+		port = 8888,
+		maxclient = max_client,
+	})
 
 	skynet.exit()
 end)

examples/watchdog.lua

@@ -1,58 +1,49 @@
 local skynet = require "skynet"
 
-local port, max_agent, buffer = ...
-local command = {}
-local agent_all = {}
+local CMD = {}
+local SOCKET = {}
 local gate
+local agent = {}
 
-function command:open(parm)
-	local fd,addr = string.match(parm,"(%d+) ([^%s]+)")
-	fd = tonumber(fd)
-	print("agent open",self,string.format("%d %d %s",self,fd,addr))
-	local client = skynet.launch("client",fd)
-	local agent = skynet.launch("snlua","agent",skynet.address(client))
-	if agent then
-		agent_all[self] = { agent , client }
-		skynet.send(gate, "text", "forward" , self, skynet.address(agent) , skynet.address(client))
+function SOCKET.open(fd, addr)
+	agent[fd] = skynet.newservice("agent")
+	skynet.call(agent[fd], "lua", "start", gate, fd)
+end
+
+local function close_agent(fd)
+	local a = agent[fd]
+	if a then
+		skynet.call(a, "lua", "exit")
+		skynet.kill(a)
+		agent[fd] = nil
 	end
 end
 
-function command:close()
-	print("agent close",self,string.format("close %d",self))
-	local agent = agent_all[self]
-	agent_all[self] = nil
-	skynet.kill(agent[1])
-	skynet.kill(agent[2])
+function SOCKET.close(fd)
+	print("socket close",fd)
+	close_agent(fd)
 end
 
-function command:data(data, session)
-	local agent = agent_all[self]
-	if agent then
-		-- PTYPE_CLIENT = 3 , read skynet.h
-		skynet.redirect(agent[1], agent[2], "client", 0, data)
-	else
-		skynet.error(string.format("agent data drop %d size=%d",self,#data))
-	end
+function SOCKET.error(fd, msg)
+	print("socket error",fd, msg)
+	close_agent(fd)
 end
 
-skynet.register_protocol {
-	name = "client",
-	id = skynet.PTYPE_CLIENT,
-}
+function CMD.start(conf)
+	skynet.call(gate, "lua", "open" , conf)
+end
 
 skynet.start(function()
-	skynet.dispatch("text", function(session, from, message)
-		local id, cmd , parm = string.match(message, "(%d+) (%w+) ?(.*)")
-		id = tonumber(id)
-		local f = command[cmd]
-		if f then
-			f(id,parm,session)
+	skynet.dispatch("lua", function(session, source, cmd, subcmd, ...)
+		if cmd == "socket" then
+			local f = SOCKET[subcmd]
+			f(...)
+			-- socket api don't need return
 		else
-			error(string.format("[watchdog] Unknown command : %s",message))
+			local f = assert(CMD[cmd])
+			skynet.ret(skynet.pack(f(subcmd, ...)))
 		end
 	end)
-	-- 0 for default client tag
-	gate = skynet.launch("gate" , "S" , skynet.address(skynet.self()), port, 0, max_agent, buffer)
-	skynet.send(gate,"text", "start")
-	skynet.register(".watchdog")
+
+	gate = skynet.newservice("gate")
 end)

service-src/service_client.c

@@ -22,7 +22,6 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t
 	if (type == PTYPE_RESPONSE) {
 		// todo: response to client with session, session may be packed into package
 	} else {
-		printf("client %d\n",type);
 		assert(type == PTYPE_TEXT);
 		// todo: support other protocol
 	}

service/gate.lua

@@ -4,13 +4,21 @@ local socketdriver = require "socketdriver"
 
 local socket
 local queue
+local watchdog
+local maxclient
+local client_number = 0
 local CMD = setmetatable({}, { __gc = function() netpack.clear(queue) end })
+local forwarding_address = {}
+local forwarding_client = {}
+local forwarding_map = {}
+local openning = {}
 
-function CMD.open( conf )
+function CMD.open( source , conf )
 	assert(not socket)
 	local address = conf.address or "0.0.0.0"
 	local port = assert(conf.port)
-	local maxclient = conf.maxclient or 1024
+	maxclient = conf.maxclient or 1024
+	watchdog = conf.watchdog or source
 	socket = socketdriver.listen(address, port)
 	socketdriver.start(socket)
 end
@@ -21,29 +29,101 @@ function CMD.close()
 	socket = nil
 end
 
+local function unforward_fd(fd)
+	local address = forwarding_address[fd]
+	if address then
+		local client = forwarding_client[fd]
+		local fm = forwarding_map[address]
+		if type(fm) == "table" then
+			fm[address] = nil
+			if next(fm) == nil then
+				forwarding_map[address] = nil
+			end
+		else
+			forwarding_map[address] = nil
+		end
+		forwarding_address[fd] = nil
+		forwarding_client[fd] = nil
+		return address, client
+	end
+end
+
+
+function CMD.forward(source, fd, client, address)
+	local addr = address or source
+	if not unforward_fd(fd) then
+		socketdriver.start(fd)
+	end
+
+	forwarding_address[fd] = addr
+	forwarding_client[fd] = client or 0
+	local fm = forwarding_map[addr]
+	if fm then
+		if type(fm) == "table" then
+			fm[addr] = true
+		else
+			fm[addr] = { [fm] = true, [addr] = true }
+		end
+	else
+		forwarding_map[addr] = fd
+	end
+end
+
+function CMD.kick(source, fd)
+	if not fd then
+		fd = forwarding_map[source]
+		if type(fd) == "table" then
+			for k in pairs(fd) do
+				socketdriver.close(k)
+			end
+			return
+		end
+	end
+	socketdriver.close(fd)
+end
+
 local MSG = {}
 
 function MSG.data(fd, msg, sz)
-	print("Data:", fd, netpack.tostring(msg, sz))
+	local address = forwarding_address[fd]
+	local client = forwarding_client[fd]
+	skynet.redirect(address, client, "client", 0, msg, sz)
 end
 
 function MSG.more()
 	for fd, msg, sz in netpack.pop, queue do
-		print("More:", fd, netpack.tostring(msg, sz))
+		local address = forwarding_address[fd]
+		local client = forwarding_client[fd]
+		skynet.redirect(address, client, "client", 0, msg, sz)
 	end
 end
 
 function MSG.open(fd, msg)
-	socketdriver.start(fd)
-	print("Open:", fd, msg)
+	if client_number >= maxclient then
+		socketdriver.close(fd)
+		return
+	end
+	openning[fd] = true
+	client_number = client_number + 1
+	skynet.send(watchdog, "lua", "socket", "open", fd, msg)
+end
+
+local function close_fd(fd)
+	local address, client = unforward_fd(fd)
+	if openning[fd] then
+		openning[fd] = nil
+		client_number = client_number - 1
+	end
 end
 
 function MSG.close(fd)
-	print("Close:", fd)
+	close_fd(fd)
+	skynet.send(watchdog, "lua", "socket", "close", fd)
 end
 
 function MSG.error(fd, msg)
-	print("Error:", fd, msg)
+	close_fd(fd)
+	skynet.send(watchdog, "lua", "socket", "error", fd)
 end
 
 skynet.register_protocol {
@@ -60,9 +140,14 @@ skynet.register_protocol {
 	end
 }
 
+skynet.register_protocol {
+	name = "client",
+	id = skynet.PTYPE_CLIENT,
+}
+
 skynet.start(function()
-	skynet.dispatch("lua", function (_,_, cmd, ...)
+	skynet.dispatch("lua", function (_, address, cmd, ...)
 		local f = assert(CMD[cmd])
-		skynet.ret(skynet.pack(f(...)))
+		skynet.ret(skynet.pack(f(address, ...)))
 	end)
 end)