add socket.response to support redis watch mode

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)
+--- watch mode
-	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
 
-	local n = tonumber(data)
+local watch = {}
-	local result = {}
+
-	local err = nil
+local watchmeta = {
-	for i = 1,n do
+	__index = watch,
-		local ok, r = read_response(fd)
+	__gc = function(self)
-		if ok then
+		self.__sock:close()
-			result[i] = r
+	end,
-		else
+}
-			if not err then
+
-				err = {}
+local function watch_login(obj, auth)
+	return function(so)
+		if auth then
+			so:request("AUTH "..auth.."\r\n", read_response)
 		end
-			table.insert(err, i .. ":" .. r)
+		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
-	if err then
-		return false, table.concat(err,",")
-	else
-		return true, result
 	end
 end
 
-function command:exec()
+function redis.watch(dbname)
-	return self[1]:request( "EXEC\r\n" , read_exec)
+	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

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

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)