redis pipeline

redis pipeline

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 > 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(cmd[1]), {cmd[2], cmd[3], cmd[4]}))
+	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,37 @@
+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)
+
+	local ret = db:pipeline {
+		{"hincrby", "hello", 1, 1},
+		{"del", "hello"},
+		{"hincrby", "hello", 3, 1},
+		{"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
+
+	db:disconnect()
+	skynet.exit()
+end)
+