Merge branch 'master' of https://github.com/gitfancode/skynet into gitfancode-master

lualib/redis.lua

@@ -4,6 +4,7 @@ local socketchannel = require "socketchannel"
 
 local table = table
 local string = string
+local assert = assert
 
 local redis = {}
 local command = {}
@@ -161,6 +162,27 @@ function command:sismember(key, value)
 	return fd:request(compose_message ("SISMEMBER", {key, value}), read_boolean)
 end
 
+function command:pipeline(ops)
+	assert(ops and #ops > 0, "pipeline is null")
+
+	local fd = self[1]
+
+	local cmds = {}
+	for _, cmd in ipairs(ops) do
+		assert(#cmd >= 2, "pipeline error, the params length is less than 2")
+		table.insert(cmds, compose_message(string.upper(table.remove(cmd, 1)), cmd))
+	end
+
+	return fd:request(table.concat(cmds, "\r\n"), function (fd)
+		local result = {}
+		for i=1, #ops do
+			local ok, out = read_response(fd)
+			table.insert(result, {ok = ok, out = out})
+		end
+		return true, result
+	end)
+end
+
 --- watch mode
 
 local watch = {}

test/testpipeline.lua

@@ -0,0 +1,72 @@
+local skynet = require "skynet"
+local redis = require "redis"
+
+local conf = {
+	host = "127.0.0.1",
+	port = 6379,
+	db = 0
+}
+
+local function read_table(t)
+    local result = { }
+    for i = 1, #t, 2 do result[t[i]] = t[i + 1] end
+    return result
+end
+
+skynet.start(function()
+	local db = redis.connect(conf)
+
+	db.pipelining = function (self, block)
+		local ops = {}
+
+		block(setmetatable({}, {
+			__index = function (_, name)
+				return function (_, ...)
+					table.insert(ops, {name, ...})
+				end
+			end
+		}))
+
+		return self:pipeline(ops)
+	end
+
+	do
+		print("test function")
+		local ret = db:pipelining(function (red)
+			red:hincrby("hello", 1, 1)
+			red:del("hello")
+			red:hmset("hello", 1, 1, 2, 2, 3, 3)
+			red:hgetall("hello")
+		end)
+
+		print(ret[1].out)
+		print(ret[2].out)
+		print(ret[3].out)
+
+		for k, v in pairs(read_table(ret[4].out)) do
+			print(k, v)
+		end
+	end
+
+	do
+		print("test table")
+		local ret = db:pipeline {
+			{"hincrby", "hello", 1, 1},
+			{"del", "hello"},
+			{"hmset", "hello", 1, 1, 2, 2, 3, 3},
+			{"hgetall", "hello"},
+		}
+
+		print(ret[1].out)
+		print(ret[2].out)
+		print(ret[3].out)
+
+		for k, v in pairs(read_table(ret[4].out)) do
+			print(k, v)
+		end
+	end
+
+	db:disconnect()
+	skynet.exit()
+end)
+