mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-25 12:43:09 +00:00
c/s json protocol example
This commit is contained in:
@@ -1,29 +1,36 @@
|
|||||||
local skynet = require "skynet"
|
local skynet = require "skynet"
|
||||||
local client
|
local jsonpack = require "jsonpack"
|
||||||
|
local netpack = require "netpack"
|
||||||
|
local socket = require "socket"
|
||||||
|
|
||||||
local CMD = {}
|
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 {
|
skynet.register_protocol {
|
||||||
name = "client",
|
name = "client",
|
||||||
id = skynet.PTYPE_CLIENT,
|
id = skynet.PTYPE_CLIENT,
|
||||||
pack = function(...) return ... end,
|
unpack = function (msg, sz)
|
||||||
unpack = skynet.tostring,
|
return jsonpack.unpack(skynet.tostring(msg,sz))
|
||||||
dispatch = function (session, address, text)
|
end,
|
||||||
-- It's client, there is no session
|
dispatch = function (_, _, session, args)
|
||||||
local result = skynet.call("SIMPLEDB", "text", text)
|
local result = skynet.call("SIMPLEDB", "lua", table.unpack(args))
|
||||||
skynet.ret(result)
|
response_client(session, result)
|
||||||
end
|
end
|
||||||
}
|
}
|
||||||
|
|
||||||
function CMD.start(gate , fd)
|
function CMD.start(gate , fd)
|
||||||
client = skynet.launch("client", fd)
|
client_fd = fd
|
||||||
skynet.call(gate, "lua", "forward", fd, client)
|
skynet.call(gate, "lua", "forward", fd)
|
||||||
skynet.send(client,"text","Welcome to skynet")
|
send_client "Welcome to skynet"
|
||||||
end
|
|
||||||
|
|
||||||
function CMD.exit()
|
|
||||||
skynet.kill(client)
|
|
||||||
client = nil
|
|
||||||
end
|
end
|
||||||
|
|
||||||
skynet.start(function()
|
skynet.start(function()
|
||||||
|
|||||||
@@ -19,16 +19,31 @@ local function dispatch()
|
|||||||
break
|
break
|
||||||
end
|
end
|
||||||
for _, v in ipairs(result) do
|
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
|
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
|
while true do
|
||||||
dispatch()
|
dispatch()
|
||||||
local cmd = socket.readline()
|
local cmd = socket.readline()
|
||||||
if cmd then
|
if cmd then
|
||||||
socket.send(fd, cmd)
|
local args = {}
|
||||||
|
string.gsub(cmd, '[^ ]+', function(v) table.insert(args, v) end )
|
||||||
|
send_request(args)
|
||||||
else
|
else
|
||||||
socket.usleep(100)
|
socket.usleep(100)
|
||||||
end
|
end
|
||||||
|
|||||||
@@ -4,34 +4,19 @@ local db = {}
|
|||||||
local command = {}
|
local command = {}
|
||||||
|
|
||||||
function command.GET(key)
|
function command.GET(key)
|
||||||
skynet.ret(db[key])
|
return db[key]
|
||||||
end
|
end
|
||||||
|
|
||||||
function command.SET(key, value)
|
function command.SET(key, value)
|
||||||
local last = db[key]
|
local last = db[key]
|
||||||
db[key] = value
|
db[key] = value
|
||||||
skynet.ret(last)
|
return last
|
||||||
end
|
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.start(function()
|
||||||
skynet.dispatch("text", function(session, address, message)
|
skynet.dispatch("lua", function(session, address, cmd, ...)
|
||||||
trace_cache[skynet.trace()] = message
|
local f = command[string.upper(cmd)]
|
||||||
local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)")
|
skynet.ret(skynet.pack(f(...)))
|
||||||
local f = command[cmd]
|
|
||||||
if f then
|
|
||||||
f(key,value)
|
|
||||||
else
|
|
||||||
skynet.ret("Invalid command : "..message)
|
|
||||||
end
|
|
||||||
end)
|
end)
|
||||||
skynet.register "SIMPLEDB"
|
skynet.register "SIMPLEDB"
|
||||||
end)
|
end)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
local skynet = require "skynet"
|
local skynet = require "skynet"
|
||||||
|
local netpack = require "netpack"
|
||||||
|
|
||||||
local CMD = {}
|
local CMD = {}
|
||||||
local SOCKET = {}
|
local SOCKET = {}
|
||||||
@@ -13,7 +14,6 @@ end
|
|||||||
local function close_agent(fd)
|
local function close_agent(fd)
|
||||||
local a = agent[fd]
|
local a = agent[fd]
|
||||||
if a then
|
if a then
|
||||||
skynet.call(a, "lua", "exit")
|
|
||||||
skynet.kill(a)
|
skynet.kill(a)
|
||||||
agent[fd] = nil
|
agent[fd] = nil
|
||||||
end
|
end
|
||||||
@@ -29,6 +29,9 @@ function SOCKET.error(fd, msg)
|
|||||||
close_agent(fd)
|
close_agent(fd)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
function SOCKET.data(fd, msg)
|
||||||
|
end
|
||||||
|
|
||||||
function CMD.start(conf)
|
function CMD.start(conf)
|
||||||
skynet.call(gate, "lua", "open" , conf)
|
skynet.call(gate, "lua", "open" , conf)
|
||||||
end
|
end
|
||||||
|
|||||||
19
lualib/jsonpack.lua
Normal file
19
lualib/jsonpack.lua
Normal file
@@ -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
|
||||||
116
service/gate.lua
116
service/gate.lua
@@ -8,10 +8,9 @@ local watchdog
|
|||||||
local maxclient
|
local maxclient
|
||||||
local client_number = 0
|
local client_number = 0
|
||||||
local CMD = setmetatable({}, { __gc = function() netpack.clear(queue) end })
|
local CMD = setmetatable({}, { __gc = function() netpack.clear(queue) end })
|
||||||
local forwarding_address = {}
|
|
||||||
local forwarding_client = {}
|
local connection = {} -- fd -> connection : { fd , client, agent , ip, mode }
|
||||||
local forwarding_map = {}
|
local forwarding = {} -- agent -> connection
|
||||||
local openning = {}
|
|
||||||
|
|
||||||
function CMD.open( source , conf )
|
function CMD.open( source , conf )
|
||||||
assert(not socket)
|
assert(not socket)
|
||||||
@@ -29,72 +28,70 @@ function CMD.close()
|
|||||||
socket = nil
|
socket = nil
|
||||||
end
|
end
|
||||||
|
|
||||||
local function unforward_fd(fd)
|
local function unforward(c)
|
||||||
local address = forwarding_address[fd]
|
if c.agent then
|
||||||
if address then
|
forwarding[c.agent] = nil
|
||||||
local client = forwarding_client[fd]
|
c.agent = nil
|
||||||
local fm = forwarding_map[address]
|
c.client = nil
|
||||||
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
|
||||||
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)
|
function CMD.forward(source, fd, client, address)
|
||||||
local addr = address or source
|
local c = assert(connection[fd])
|
||||||
if not unforward_fd(fd) then
|
unforward(c)
|
||||||
socketdriver.start(fd)
|
start(c)
|
||||||
end
|
|
||||||
|
|
||||||
forwarding_address[fd] = addr
|
c.client = client or 0
|
||||||
forwarding_client[fd] = client or 0
|
c.agent = address or source
|
||||||
local fm = forwarding_map[addr]
|
|
||||||
if fm then
|
forwarding[c.agent] = c
|
||||||
if type(fm) == "table" then
|
end
|
||||||
fm[addr] = true
|
|
||||||
else
|
function CMD.accept(source, fd)
|
||||||
fm[addr] = { [fm] = true, [addr] = true }
|
local c = assert(connection[fd])
|
||||||
end
|
unforward(c)
|
||||||
else
|
start(c)
|
||||||
forwarding_map[addr] = fd
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
|
|
||||||
function CMD.kick(source, fd)
|
function CMD.kick(source, fd)
|
||||||
if not fd then
|
local c
|
||||||
fd = forwarding_map[source]
|
if fd then
|
||||||
if type(fd) == "table" then
|
c = connection[fd]
|
||||||
for k in pairs(fd) do
|
else
|
||||||
socketdriver.close(k)
|
c = forwarding[source]
|
||||||
end
|
end
|
||||||
return
|
|
||||||
end
|
assert(c)
|
||||||
|
|
||||||
|
if c.mode ~= "close" then
|
||||||
|
c.mode = "close"
|
||||||
|
socketdriver.close(c.fd)
|
||||||
end
|
end
|
||||||
socketdriver.close(fd)
|
|
||||||
end
|
end
|
||||||
|
|
||||||
local MSG = {}
|
local MSG = {}
|
||||||
|
|
||||||
function MSG.data(fd, msg, sz)
|
function MSG.data(fd, msg, sz)
|
||||||
local address = forwarding_address[fd]
|
-- recv a package, forward it
|
||||||
local client = forwarding_client[fd]
|
local c = connection[fd]
|
||||||
skynet.redirect(address, client, "client", 0, msg, sz)
|
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
|
end
|
||||||
|
|
||||||
function MSG.more()
|
function MSG.more()
|
||||||
for fd, msg, sz in netpack.pop, queue do
|
for fd, msg, sz in netpack.pop, queue do
|
||||||
local address = forwarding_address[fd]
|
MSG.data(fd, msg, sz)
|
||||||
local client = forwarding_client[fd]
|
|
||||||
skynet.redirect(address, client, "client", 0, msg, sz)
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
@@ -103,15 +100,20 @@ function MSG.open(fd, msg)
|
|||||||
socketdriver.close(fd)
|
socketdriver.close(fd)
|
||||||
return
|
return
|
||||||
end
|
end
|
||||||
openning[fd] = true
|
local c = {
|
||||||
|
fd = fd,
|
||||||
|
ip = msg,
|
||||||
|
}
|
||||||
|
connection[fd] = c
|
||||||
client_number = client_number + 1
|
client_number = client_number + 1
|
||||||
skynet.send(watchdog, "lua", "socket", "open", fd, msg)
|
skynet.send(watchdog, "lua", "socket", "open", fd, msg)
|
||||||
end
|
end
|
||||||
|
|
||||||
local function close_fd(fd)
|
local function close_fd(fd, message)
|
||||||
local address, client = unforward_fd(fd)
|
local c = connection[fd]
|
||||||
if openning[fd] then
|
if c then
|
||||||
openning[fd] = nil
|
unforward(c)
|
||||||
|
connection[fd] = nil
|
||||||
client_number = client_number - 1
|
client_number = client_number - 1
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
@@ -123,7 +125,7 @@ end
|
|||||||
|
|
||||||
function MSG.error(fd, msg)
|
function MSG.error(fd, msg)
|
||||||
close_fd(fd)
|
close_fd(fd)
|
||||||
skynet.send(watchdog, "lua", "socket", "error", fd)
|
skynet.send(watchdog, "lua", "socket", "error", fd, msg)
|
||||||
end
|
end
|
||||||
|
|
||||||
skynet.register_protocol {
|
skynet.register_protocol {
|
||||||
|
|||||||
Reference in New Issue
Block a user