You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
110 lines
2.8 KiB
Lua
110 lines
2.8 KiB
Lua
local skynet = require "skynet"
|
|
local cjson = require "cjson"
|
|
local dateutils = require "dateutils"
|
|
local redis = require "skynet.db.redis"
|
|
local settings = require "settings"
|
|
local keysutils = require "keysutils"
|
|
local skynet_node_name = ...
|
|
|
|
require "functions"
|
|
|
|
require "skynet.manager"
|
|
|
|
local CMD = {}
|
|
local lasttime
|
|
local db
|
|
local agents = {}
|
|
|
|
|
|
function CMD.online(uid, agent)
|
|
agents[uid] = agent
|
|
return "ok"
|
|
end
|
|
|
|
function CMD.offline(uid)
|
|
if agents[uid] then
|
|
agents[uid] = nil
|
|
end
|
|
return "ok"
|
|
end
|
|
|
|
local function func(data)
|
|
local order_no = data.order_no
|
|
local uid = data.uid
|
|
if agents[uid] then
|
|
local agent = agents[uid]
|
|
DEBUG("func agent = ", agent)
|
|
local ok = skynet.call(agent, "lua", "recharge_callback", order_no)
|
|
else
|
|
-- assert(false, "运行到这里就错了~~~~")
|
|
end
|
|
return "ok"
|
|
end
|
|
|
|
|
|
local function lpoprpush(queue, queuebakup)
|
|
local script = [[
|
|
local v = redis.call('lpop', ARGV[1])
|
|
if not v then
|
|
return nil
|
|
end
|
|
redis.call('rpush', ARGV[2], v)
|
|
return v
|
|
]]
|
|
local t = db:eval(script, 2, "k2", "k1", queuebakup, queue)
|
|
return t
|
|
end
|
|
|
|
local function copyback(queue, queuebakup)
|
|
local t = lpoprpush(queue, queuebakup)
|
|
while t do
|
|
t = lpoprpush(queue, queuebakup)
|
|
end
|
|
end
|
|
|
|
skynet.start(function()
|
|
skynet.dispatch("lua", function(session, source, cmd, ...)
|
|
local f = assert(CMD[cmd])
|
|
skynet.ret(skynet.pack(f(...)))
|
|
end)
|
|
local cnf = settings.db_cnf[skynet_node_name].redismq_cnf
|
|
|
|
local ok, d = pcall(redis.connect, cnf)
|
|
if ok then
|
|
db = d
|
|
else
|
|
ERROR("---order redis connect error---", inspect(cnf) )
|
|
end
|
|
local k1 = "leshusanguo:pay:queue"
|
|
local k2 = "leshusanguo:pay:queue_backup"
|
|
copyback(k1, k2)
|
|
|
|
skynet.fork(function()
|
|
while true do
|
|
if db then
|
|
local res = db:brpoplpush(k1, k2, 10)
|
|
-- DEBUG("order brpoplpush res = ", res)
|
|
if res then
|
|
local ok, ret = pcall(cjson.decode, res)
|
|
if ok then
|
|
INFO("order brpoplpush ret = ", DUMP(ret))
|
|
local r = func(ret)
|
|
if not r then
|
|
ERROR("order error = ", DUMP(r))
|
|
else
|
|
db:lrem(k2, -1, res)
|
|
end
|
|
end
|
|
end
|
|
if not lasttime or lasttime < os.time() - 10 then
|
|
lasttime = os.time()
|
|
db:ping()
|
|
end
|
|
else
|
|
skynet.sleep(1000)
|
|
end
|
|
end
|
|
end)
|
|
skynet.register('.' .. SERVICE_NAME)
|
|
end)
|