add gateserver

This commit is contained in:
Cloud Wu
2014-07-13 14:54:53 +08:00
parent 0dba8ed385
commit e1674f04c3
3 changed files with 58 additions and 114 deletions

View File

@@ -11,6 +11,7 @@ skynet.start(function()
skynet.call(watchdog, "lua", "start", { skynet.call(watchdog, "lua", "start", {
port = 8888, port = 8888,
maxclient = max_client, maxclient = max_client,
nodelay = true,
}) })
print("Watchdog listen on ", 8888) print("Watchdog listen on ", 8888)

View File

@@ -33,7 +33,6 @@ function SOCKET.data(fd, msg)
end end
function CMD.start(conf) function CMD.start(conf)
skynet.call(gate, "lua", "nodelay", true)
skynet.call(gate, "lua", "open" , conf) skynet.call(gate, "lua", "open" , conf)
end end

View File

@@ -1,39 +1,41 @@
local skynet = require "skynet" local skynet = require "skynet"
local gateserver = require "gateserver"
local netpack = require "netpack" local netpack = require "netpack"
local socketdriver = require "socketdriver"
local socket
local queue
local watchdog local watchdog
local maxclient
local client_number = 0
local CMD = setmetatable({}, { __gc = function() netpack.clear(queue) end })
local nodelay = false
local connection = {} -- fd -> connection : { fd , client, agent , ip, mode } local connection = {} -- fd -> connection : { fd , client, agent , ip, mode }
local forwarding = {} -- agent -> connection local forwarding = {} -- agent -> connection
function CMD.open( source , conf ) skynet.register_protocol {
assert(not socket) name = "client",
local address = conf.address or "0.0.0.0" id = skynet.PTYPE_CLIENT,
local port = assert(conf.port) }
maxclient = conf.maxclient or 1024
local handler = {}
function handler.open(source, conf)
watchdog = conf.watchdog or source watchdog = conf.watchdog or source
socket = socketdriver.listen(address, port) maxclient = conf.maxclient or 1024
socketdriver.start(socket)
end end
function CMD.nodelay(source, v) function handler.message(fd, msg, sz)
if v ~= false then -- recv a package, forward it
v = true 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", fd, netpack.tostring(msg, sz))
end end
nodelay = v
end end
function CMD.close() function handler.connect(fd, addr)
assert(socket) local c = {
socketdriver.close(socket) fd = fd,
socket = nil ip = msg,
}
connection[fd] = c
skynet.send(watchdog, "lua", "socket", "open", fd, addr)
end end
local function unforward(c) local function unforward(c)
@@ -44,28 +46,47 @@ local function unforward(c)
end end
end end
local function start(c) local function close_fd(fd)
if not c.mode then local c = connection[fd]
c.mode = "open" if c then
socketdriver.start(c.fd) unforward(c)
connection[fd] = nil
end
end
function handler.close(fd)
close_fd(fd)
skynet.send(watchdog, "lua", "socket", "close", fd)
end
function handler.error(fd, msg)
close_fd(fd)
skynet.send(watchdog, "lua", "socket", "error", fd, msg)
end
local CMD = {}
local function unforward(c)
if c.agent then
forwarding[c.agent] = nil
c.agent = nil
c.client = nil
end end
end end
function CMD.forward(source, fd, client, address) function CMD.forward(source, fd, client, address)
local c = assert(connection[fd]) local c = assert(connection[fd])
unforward(c) unforward(c)
start(c)
c.client = client or 0 c.client = client or 0
c.agent = address or source c.agent = address or source
forwarding[c.agent] = c forwarding[c.agent] = c
gateserver.openclient(fd)
end end
function CMD.accept(source, fd) function CMD.accept(source, fd)
local c = assert(connection[fd]) local c = assert(connection[fd])
unforward(c) unforward(c)
start(c) gateserver.openclient(fd)
end end
function CMD.kick(source, fd) function CMD.kick(source, fd)
@@ -78,89 +99,12 @@ function CMD.kick(source, fd)
assert(c) assert(c)
if c.mode ~= "close" then gateserver.closeclient(fd)
c.mode = "close"
socketdriver.close(c.fd)
end
end end
local MSG = {} function handler.command(cmd, source, ...)
local f = assert(CMD[cmd])
function MSG.data(fd, msg, sz) return f(source, ...)
-- 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", fd, netpack.tostring(msg, sz))
end
end end
function MSG.more() gateserver.start(handler)
for fd, msg, sz in netpack.pop, queue do
MSG.data(fd, msg, sz)
end
end
function MSG.open(fd, msg)
if client_number >= maxclient then
socketdriver.close(fd)
return
end
local c = {
fd = fd,
ip = msg,
}
connection[fd] = c
client_number = client_number + 1
if nodelay then
socketdriver.nodelay(fd)
end
skynet.send(watchdog, "lua", "socket", "open", fd, msg)
end
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
function MSG.close(fd)
close_fd(fd)
skynet.send(watchdog, "lua", "socket", "close", fd)
end
function MSG.error(fd, msg)
close_fd(fd)
skynet.send(watchdog, "lua", "socket", "error", fd, msg)
end
skynet.register_protocol {
name = "socket",
id = skynet.PTYPE_SOCKET, -- PTYPE_SOCKET = 6
unpack = function ( msg, sz )
return netpack.filter( queue, msg, sz)
end,
dispatch = function (_, _, q, type, ...)
queue = q
if type then
MSG[type](...)
end
end
}
skynet.register_protocol {
name = "client",
id = skynet.PTYPE_CLIENT,
}
skynet.start(function()
skynet.dispatch("lua", function (_, address, cmd, ...)
local f = assert(CMD[cmd])
skynet.ret(skynet.pack(f(address, ...)))
end)
end)