diff --git a/examples/agent.lua b/examples/agent.lua index 69962315..3169e324 100644 --- a/examples/agent.lua +++ b/examples/agent.lua @@ -1,29 +1,36 @@ local skynet = require "skynet" -local client +local jsonpack = require "jsonpack" +local netpack = require "netpack" +local socket = require "socket" local CMD = {} +local client_fd + +local function send_client(v) + socket.write(client_fd, netpack.pack(jsonpack.pack(0,v))) +end + +local function response_client(session,v) + socket.write(client_fd, netpack.pack(jsonpack.response(session,v))) +end + skynet.register_protocol { name = "client", id = skynet.PTYPE_CLIENT, - pack = function(...) return ... end, - unpack = skynet.tostring, - dispatch = function (session, address, text) - -- It's client, there is no session - local result = skynet.call("SIMPLEDB", "text", text) - skynet.ret(result) + unpack = function (msg, sz) + return jsonpack.unpack(skynet.tostring(msg,sz)) + end, + dispatch = function (_, _, session, args) + local result = skynet.call("SIMPLEDB", "lua", table.unpack(args)) + response_client(session, result) end } 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 + client_fd = fd + skynet.call(gate, "lua", "forward", fd) + send_client "Welcome to skynet" end skynet.start(function() diff --git a/examples/client.lua b/examples/client.lua index 7bfdc8b5..b50d0039 100644 --- a/examples/client.lua +++ b/examples/client.lua @@ -19,16 +19,31 @@ local function dispatch() break end for _, v in ipairs(result) do - print(v) + local session,t,str = string.match(v, "(%d+)(.)(.*)") + assert(t == '-' or t == '+') + session = tonumber(session) + local result = cjson.decode(str) + print("Response:",session, result) end end end +local session = 0 + +local function send_request(v) + session = session + 1 + local str = string.format("%d+%s",session, cjson.encode(v)) + socket.send(fd, str) + print("Request:", session) +end + while true do dispatch() local cmd = socket.readline() if cmd then - socket.send(fd, cmd) + local args = {} + string.gsub(cmd, '[^ ]+', function(v) table.insert(args, v) end ) + send_request(args) else socket.usleep(100) end diff --git a/examples/simpledb.lua b/examples/simpledb.lua index 39d6ac06..a11fd8c5 100644 --- a/examples/simpledb.lua +++ b/examples/simpledb.lua @@ -4,34 +4,19 @@ local db = {} local command = {} function command.GET(key) - skynet.ret(db[key]) + return db[key] end function command.SET(key, value) local last = db[key] db[key] = value - skynet.ret(last) + return last end -local trace_cache = {} - -skynet.trace_callback(function(handle, ti) --- print("TRACE",trace_cache[handle],ti) - trace_cache[handle] = nil -end) - skynet.start(function() - skynet.dispatch("text", function(session, address, message) - trace_cache[skynet.trace()] = message - local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)") - local f = command[cmd] - if f then - f(key,value) - else - skynet.ret("Invalid command : "..message) - end + skynet.dispatch("lua", function(session, address, cmd, ...) + local f = command[string.upper(cmd)] + skynet.ret(skynet.pack(f(...))) end) skynet.register "SIMPLEDB" end) - - diff --git a/examples/watchdog.lua b/examples/watchdog.lua index a6d5339a..8ead2253 100644 --- a/examples/watchdog.lua +++ b/examples/watchdog.lua @@ -1,4 +1,5 @@ local skynet = require "skynet" +local netpack = require "netpack" local CMD = {} local SOCKET = {} @@ -13,7 +14,6 @@ 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 @@ -29,6 +29,9 @@ function SOCKET.error(fd, msg) close_agent(fd) end +function SOCKET.data(fd, msg) +end + function CMD.start(conf) skynet.call(gate, "lua", "open" , conf) end diff --git a/lualib/jsonpack.lua b/lualib/jsonpack.lua new file mode 100644 index 00000000..b1da80cd --- /dev/null +++ b/lualib/jsonpack.lua @@ -0,0 +1,19 @@ +local cjson = require "cjson" + +local jsonpack = {} + +function jsonpack.pack(session, v) + return string.format("%d+%s", session, cjson.encode(v)) +end + +function jsonpack.response(session, v) + return string.format("%d-%s",session, cjson.encode(v)) +end + +function jsonpack.unpack(msg) + local session,t,str = string.match(msg, "(%d+)(.)(.*)") + assert(t == '+') + return tonumber(session) , cjson.decode(str) +end + +return jsonpack \ No newline at end of file diff --git a/service/gate.lua b/service/gate.lua index 86d00209..1f3aea0b 100644 --- a/service/gate.lua +++ b/service/gate.lua @@ -8,10 +8,9 @@ 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 = {} + +local connection = {} -- fd -> connection : { fd , client, agent , ip, mode } +local forwarding = {} -- agent -> connection function CMD.open( source , conf ) assert(not socket) @@ -29,72 +28,70 @@ 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 +local function unforward(c) + if c.agent then + forwarding[c.agent] = nil + c.agent = nil + c.client = nil end end +local function start(c) + if not c.mode then + c.mode = "open" + socketdriver.start(c.fd) + end +end function CMD.forward(source, fd, client, address) - local addr = address or source - if not unforward_fd(fd) then - socketdriver.start(fd) - end + local c = assert(connection[fd]) + unforward(c) + start(c) - 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 + c.client = client or 0 + c.agent = address or source + + forwarding[c.agent] = c +end + +function CMD.accept(source, fd) + local c = assert(connection[fd]) + unforward(c) + start(c) 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 + local c + if fd then + c = connection[fd] + else + c = forwarding[source] + end + + assert(c) + + if c.mode ~= "close" then + c.mode = "close" + socketdriver.close(c.fd) end - socketdriver.close(fd) end local MSG = {} function MSG.data(fd, msg, sz) - local address = forwarding_address[fd] - local client = forwarding_client[fd] - skynet.redirect(address, client, "client", 0, msg, sz) + -- recv a package, forward it + local c = connection[fd] + local agent = c.agent + if agent then + skynet.redirect(agent, c.client, "client", 0, msg, sz) + else + skynet.send(watchdog, "lua", "socket", "data", netpack.tostring(msg, sz)) + end end function MSG.more() for fd, msg, sz in netpack.pop, queue do - local address = forwarding_address[fd] - local client = forwarding_client[fd] - skynet.redirect(address, client, "client", 0, msg, sz) + MSG.data(fd, msg, sz) end end @@ -103,15 +100,20 @@ function MSG.open(fd, msg) socketdriver.close(fd) return end - openning[fd] = true + local c = { + fd = fd, + ip = msg, + } + connection[fd] = c 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 +local function close_fd(fd, message) + local c = connection[fd] + if c then + unforward(c) + connection[fd] = nil client_number = client_number - 1 end end @@ -123,7 +125,7 @@ end function MSG.error(fd, msg) close_fd(fd) - skynet.send(watchdog, "lua", "socket", "error", fd) + skynet.send(watchdog, "lua", "socket", "error", fd, msg) end skynet.register_protocol {