From ffbc22e088e4d917343334eecf03309debf8e695 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Sun, 23 Mar 2014 19:46:21 +0800 Subject: [PATCH] add socket.response to support redis watch mode --- lualib/redis.lua | 139 +++++++++++++++++++++++++++++------------- lualib/socket.lua | 31 +++++++++- service/testredis.lua | 28 +++++++-- 3 files changed, 149 insertions(+), 49 deletions(-) diff --git a/lualib/redis.lua b/lualib/redis.lua index efa0cd20..7142728d 100644 --- a/lualib/redis.lua +++ b/lualib/redis.lua @@ -19,24 +19,6 @@ local meta = { ---------- redis response local redcmd = {} -redcmd[42] = function(fd, data) -- '*' - local n = tonumber(data) - if n < 0 then - return true, nil - end - local bulk = {} - for i = 1,n do - local line = fd:readline "\r\n" - local bytes = tonumber(string.sub(line,2)) - if bytes >= 0 then - local data = fd:read(bytes + 2) - -- bulk[i] = nil when bytes < 0 - bulk[i] = string.sub(data,1,-3) - end - end - return true, bulk -end - redcmd[36] = function(fd, data) -- '$' local bytes = tonumber(data) if bytes < 0 then @@ -65,6 +47,25 @@ local function read_response(fd) local data = string.sub(result,2) return redcmd[firstchar](fd,data) end + +redcmd[42] = function(fd, data) -- '*' + local n = tonumber(data) + if n < 0 then + return true, nil + end + local bulk = {} + local noerr = true + for i = 1,n do + local ok, v = read_response(fd) + if ok then + bulk[i] = v + else + noerr = false + end + end + return noerr, bulk +end + ------------------- local function redis_login(auth, db) @@ -141,37 +142,87 @@ function command:sismember(key, value) return fd:request(compose_message { "SISMEMBER", key, value }, read_boolean) end -local function read_exec(fd) - local result = fd:readline "\r\n" - local firstchar = string.byte(result) - local data = string.sub(result,2) - if firstchar ~= 42 then - return false, data - end +--- watch mode - local n = tonumber(data) - local result = {} - local err = nil - for i = 1,n do - local ok, r = read_response(fd) - if ok then - result[i] = r - else - if not err then - err = {} - end - table.insert(err, i .. ":" .. r) +local watch = {} + +local watchmeta = { + __index = watch, + __gc = function(self) + self.__sock:close() + end, +} + +local function watch_login(obj, auth) + return function(so) + if auth then + so:request("AUTH "..auth.."\r\n", read_response) + end + for k in pairs(obj.__psubscribe) do + so:request(compose_message { "PSUBSCRIBE", k }) + end + for k in pairs(obj.__subscribe) do + so:request(compose_message { "SUBSCRIBE", k }) end - end - if err then - return false, table.concat(err,",") - else - return true, result end end -function command:exec() - return self[1]:request( "EXEC\r\n" , read_exec) +function redis.watch(dbname) + local db_conf = name[dbname] + local obj = { + __subscribe = {}, + __psubscribe = {}, + } + local channel = socket.channel { + host = db_conf.host, + port = db_conf.port or 6379, + auth = watch_login(obj, db_conf.auth), + } + obj.__sock = channel + + return setmetatable( obj, watchmeta ) +end + +function watch:disconnect() + self.__sock:close() + setmetatable(self, nil) +end + +local function watch_func( name ) + local NAME = string.upper(name) + watch[name] = function(self, ...) + local so = self.__sock + for i = 1, select("#", ...) do + local v = select(i, ...) + so:request(compose_message { NAME, v }) + end + end +end + +watch_func "subscribe" +watch_func "psubscribe" +watch_func "unsubscribe" +watch_func "punsubscribe" + +function watch:message() + local so = self.__sock + while true do + local ret = so:response(read_response) + local type , channel, data , data2 = ret[1], ret[2], ret[3], ret[4] + if type == "message" then + return data, channel + elseif type == "pmessage" then + return data2, data, channel + elseif type == "subscribe" then + self.__subscribe[channel] = true + elseif type == "psubscribe" then + self.__psubscribe[channel] = true + elseif type == "unsubscribe" then + self.__subscribe[channel] = nil + elseif type == "punsubscribe" then + self.__psubscribe[channel] = nil + end + end end return redis diff --git a/lualib/socket.lua b/lualib/socket.lua index db2a9912..7f3dad67 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -366,7 +366,7 @@ local function dispatch_response(self) -- response() return session while self.__sock do local ok , session, result_ok, result_data = pcall(response, self.__sock) - if ok then + if ok and session then local co = self.__thread[session] self.__thread[session] = nil if co then @@ -530,6 +530,35 @@ function channel:request(request, response) end end +function channel:response(response) + if not self.__sock then + assert(not self.__closed) + reconnect_channel(self) + end + assert(type(response) == "function") + + local co = coroutine.running() + table.insert(self.__request, response) + table.insert(self.__thread, co) + + skynet.wait() + + local result = self.__result[co] + self.__result[co] = nil + local result_data = self.__result_data[co] + self.__result_data[co] = nil + + if result == socket_error then + if result_data then + print("socket: dispatch", request, result_data) + end + return self:response(response) + else + assert(result, result_data) + return result_data + end +end + function channel:close() if not self.__closed then self.__closed = true diff --git a/service/testredis.lua b/service/testredis.lua index 70f2e2a1..5da19729 100644 --- a/service/testredis.lua +++ b/service/testredis.lua @@ -1,8 +1,19 @@ local skynet = require "skynet" local redis = require "redis" +local function watching() + local w = redis.watch "main" + w:subscribe "foo" + w:psubscribe "hello.*" + while true do + print("Watch", w:message()) + end +end + skynet.start(function() + skynet.fork(watching) local db = redis.connect "main" + db:del "C" db:set("A", "hello") db:set("B", "world") @@ -12,7 +23,7 @@ skynet.start(function() print(db:get("B")) db:del "D" - for i=1,1000 do + for i=1,10 do db:hset("D",i,i) end local r = db:hvals "D" @@ -20,7 +31,6 @@ skynet.start(function() print(k,v) end ---[[ db:multi() db:get "A" db:get "B" @@ -28,14 +38,24 @@ skynet.start(function() for k,v in ipairs(t) do print("Exec", v) end -]] + print(db:exists "A") print(db:get "A") print(db:set("A","hello world")) print(db:get("A")) print(db:sismember("C","one")) print(db:sismember("C","two")) + + print("===========publish============") + + for i=1,10 do + db:publish("foo", i) + end + for i=11,20 do + db:publish("hello.foo", i) + end + db:disconnect() - skynet.exit() +-- skynet.exit() end)