From 471aecd708156f922ef9a08235b96d2c7559fb96 Mon Sep 17 00:00:00 2001 From: xiefan Date: Thu, 12 Nov 2015 20:45:19 +0800 Subject: [PATCH 1/3] redis pipeline redis pipeline --- lualib/redis.lua | 22 ++++++++++++++++++++++ test/testpipeline.lua | 37 +++++++++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+) create mode 100644 test/testpipeline.lua diff --git a/lualib/redis.lua b/lualib/redis.lua index b1e44185..c05e8423 100644 --- a/lualib/redis.lua +++ b/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 = {} diff --git a/test/testpipeline.lua b/test/testpipeline.lua new file mode 100644 index 00000000..ae55b766 --- /dev/null +++ b/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) + From d2396f9d204e89f95dea2782ce67a613b878a8c6 Mon Sep 17 00:00:00 2001 From: xiefan Date: Fri, 13 Nov 2015 08:10:33 +0800 Subject: [PATCH 2/3] Revert "redis pipeline" This reverts commit 471aecd708156f922ef9a08235b96d2c7559fb96. --- lualib/redis.lua | 22 ---------------------- test/testpipeline.lua | 37 ------------------------------------- 2 files changed, 59 deletions(-) delete mode 100644 test/testpipeline.lua diff --git a/lualib/redis.lua b/lualib/redis.lua index c05e8423..b1e44185 100644 --- a/lualib/redis.lua +++ b/lualib/redis.lua @@ -4,7 +4,6 @@ local socketchannel = require "socketchannel" local table = table local string = string -local assert = assert local redis = {} local command = {} @@ -162,27 +161,6 @@ 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 = {} diff --git a/test/testpipeline.lua b/test/testpipeline.lua deleted file mode 100644 index ae55b766..00000000 --- a/test/testpipeline.lua +++ /dev/null @@ -1,37 +0,0 @@ -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) - From 34d9b5555df178456776b6a023c4883dac95fa5e Mon Sep 17 00:00:00 2001 From: xiefan Date: Fri, 13 Nov 2015 08:16:47 +0800 Subject: [PATCH 3/3] reds pipeline v2(support function as pipeline params) fix a bug & add function as pipeline params --- lualib/redis.lua | 22 +++++++++++++ test/testpipeline.lua | 72 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 94 insertions(+) create mode 100644 test/testpipeline.lua diff --git a/lualib/redis.lua b/lualib/redis.lua index b1e44185..2006c335 100644 --- a/lualib/redis.lua +++ b/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 = {} diff --git a/test/testpipeline.lua b/test/testpipeline.lua new file mode 100644 index 00000000..4b0ecafb --- /dev/null +++ b/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) +