mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-24 12:20:41 +00:00
rewrite redis driver , avoid table.concat
This commit is contained in:
@@ -97,41 +97,48 @@ function command:disconnect()
|
|||||||
end
|
end
|
||||||
|
|
||||||
-- msg could be any type of value
|
-- msg could be any type of value
|
||||||
local function pack_value(lines, v)
|
|
||||||
if v == nil then
|
|
||||||
return
|
|
||||||
end
|
|
||||||
|
|
||||||
v = tostring(v)
|
local header_cache = setmetatable({}, {
|
||||||
|
__mode = "kv",
|
||||||
|
__index = function(t,k)
|
||||||
|
local s = "\r\n$" .. k .. "\r\n"
|
||||||
|
t[k] = s
|
||||||
|
return s
|
||||||
|
end})
|
||||||
|
|
||||||
table.insert(lines,"$"..#v)
|
local command_cache = setmetatable({}, {
|
||||||
table.insert(lines,v)
|
__mode = "kv",
|
||||||
end
|
__index = function(t,cmd)
|
||||||
|
local s = "\r\n$"..#cmd.."\r\n"..cmd:upper()
|
||||||
|
t[cmd] = s
|
||||||
|
return s
|
||||||
|
end})
|
||||||
|
|
||||||
local function compose_message(cmd, msg)
|
local function compose_message(cmd, msg)
|
||||||
local len = 1
|
|
||||||
local t = type(msg)
|
local t = type(msg)
|
||||||
|
local lines = {}
|
||||||
|
|
||||||
if t == "table" then
|
if t == "table" then
|
||||||
len = len + #msg
|
lines[1] = "*" .. (#msg+1)
|
||||||
elseif t ~= nil then
|
lines[2] = command_cache[cmd]
|
||||||
len = len + 1
|
local idx = 3
|
||||||
end
|
|
||||||
|
|
||||||
local lines = {"*" .. len}
|
|
||||||
pack_value(lines, cmd)
|
|
||||||
|
|
||||||
if t == "table" then
|
|
||||||
for _,v in ipairs(msg) do
|
for _,v in ipairs(msg) do
|
||||||
pack_value(lines, v)
|
v= tostring(v)
|
||||||
|
lines[idx] = header_cache[#v]
|
||||||
|
lines[idx+1] = v
|
||||||
|
idx = idx + 2
|
||||||
end
|
end
|
||||||
|
lines[idx] = "\r\n"
|
||||||
else
|
else
|
||||||
pack_value(lines, msg)
|
msg = tostring(msg)
|
||||||
|
lines[1] = "*2"
|
||||||
|
lines[2] = command_cache[cmd]
|
||||||
|
lines[3] = header_cache[#msg]
|
||||||
|
lines[4] = msg
|
||||||
|
lines[5] = "\r\n"
|
||||||
end
|
end
|
||||||
table.insert(lines, "")
|
|
||||||
|
|
||||||
local chunk = table.concat(lines,"\r\n")
|
return lines
|
||||||
return chunk
|
|
||||||
end
|
end
|
||||||
|
|
||||||
setmetatable(command, { __index = function(t,k)
|
setmetatable(command, { __index = function(t,k)
|
||||||
@@ -162,25 +169,46 @@ function command:sismember(key, value)
|
|||||||
return fd:request(compose_message ("SISMEMBER", {key, value}), read_boolean)
|
return fd:request(compose_message ("SISMEMBER", {key, value}), read_boolean)
|
||||||
end
|
end
|
||||||
|
|
||||||
function command:pipeline(ops)
|
local function compose_table(lines, msg)
|
||||||
|
local tinsert = table.insert
|
||||||
|
tinsert(lines, "*" .. #msg)
|
||||||
|
for _,v in ipairs(msg) do
|
||||||
|
v = tostring(v)
|
||||||
|
tinsert(lines,header_cache[#v])
|
||||||
|
tinsert(lines,v)
|
||||||
|
end
|
||||||
|
tinsert(lines, "\r\n")
|
||||||
|
return lines
|
||||||
|
end
|
||||||
|
|
||||||
|
function command:pipeline(ops,resp)
|
||||||
assert(ops and #ops > 0, "pipeline is null")
|
assert(ops and #ops > 0, "pipeline is null")
|
||||||
|
|
||||||
local fd = self[1]
|
local fd = self[1]
|
||||||
|
|
||||||
local cmds = {}
|
local cmds = {}
|
||||||
for _, cmd in ipairs(ops) do
|
for _, cmd in ipairs(ops) do
|
||||||
assert(#cmd >= 2, "pipeline error, the params length is less than 2")
|
compose_table(cmds, cmd)
|
||||||
table.insert(cmds, compose_message(string.upper(table.remove(cmd, 1)), cmd))
|
|
||||||
end
|
end
|
||||||
|
|
||||||
return fd:request(table.concat(cmds, "\r\n"), function (fd)
|
if resp then
|
||||||
local result = {}
|
return fd:request(cmds, function (fd)
|
||||||
for i=1, #ops do
|
for i=1, #ops do
|
||||||
local ok, out = read_response(fd)
|
local ok, out = read_response(fd)
|
||||||
table.insert(result, {ok = ok, out = out})
|
table.insert(resp, {ok = ok, out = out})
|
||||||
end
|
end
|
||||||
return true, result
|
return true, resp
|
||||||
end)
|
end)
|
||||||
|
else
|
||||||
|
return fd:request(cmds, function (fd)
|
||||||
|
local ok, out
|
||||||
|
for i=1, #ops do
|
||||||
|
ok, out = read_response(fd)
|
||||||
|
end
|
||||||
|
-- return last response
|
||||||
|
return ok,out
|
||||||
|
end)
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
--- watch mode
|
--- watch mode
|
||||||
|
|||||||
@@ -33,29 +33,27 @@ skynet.start(function()
|
|||||||
do
|
do
|
||||||
print("test function")
|
print("test function")
|
||||||
local ret = db:pipelining(function (red)
|
local ret = db:pipelining(function (red)
|
||||||
|
red:multi()
|
||||||
red:hincrby("hello", 1, 1)
|
red:hincrby("hello", 1, 1)
|
||||||
red:del("hello")
|
red:del("hello")
|
||||||
red:hmset("hello", 1, 1, 2, 2, 3, 3)
|
red:hmset("hello", 1, 1, 2, 2, 3, 3)
|
||||||
red:hgetall("hello")
|
red:hgetall("hello")
|
||||||
|
red:exec()
|
||||||
end)
|
end)
|
||||||
|
-- ret is the result of red:exec()
|
||||||
print(ret[1].out)
|
for k, v in pairs(read_table(ret[4])) do
|
||||||
print(ret[2].out)
|
|
||||||
print(ret[3].out)
|
|
||||||
|
|
||||||
for k, v in pairs(read_table(ret[4].out)) do
|
|
||||||
print(k, v)
|
print(k, v)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
do
|
do
|
||||||
print("test table")
|
print("test table")
|
||||||
local ret = db:pipeline {
|
local ret = db:pipeline({
|
||||||
{"hincrby", "hello", 1, 1},
|
{"hincrby", "hello", 1, 1},
|
||||||
{"del", "hello"},
|
{"del", "hello"},
|
||||||
{"hmset", "hello", 1, 1, 2, 2, 3, 3},
|
{"hmset", "hello", 1, 1, 2, 2, 3, 3},
|
||||||
{"hgetall", "hello"},
|
{"hgetall", "hello"},
|
||||||
}
|
}, {}) -- offer a {} for result
|
||||||
|
|
||||||
print(ret[1].out)
|
print(ret[1].out)
|
||||||
print(ret[2].out)
|
print(ret[2].out)
|
||||||
|
|||||||
Reference in New Issue
Block a user