bugfix: handle mixture of read_reply and other commands. (#195)

Author: spacewander

lib/resty/redis.lua

@@ -3,6 +3,8 @@
 
 local sub = string.sub
 local byte = string.byte
+local tab_insert = table.insert
+local tab_remove = table.remove
 local tcp = ngx.socket.tcp
 local null = ngx.null
 local type = type
@@ -306,6 +308,12 @@ local function _gen_req(args)
 end
 
 
+local function _check_msg(self, res)
+    return rawget(self, "_subscribed") and
+        type(res) == "table" and res[1] == "message"
+end
+
+
 local function _do_cmd(self, ...)
     local args = {...}
 
@@ -329,7 +337,17 @@ local function _do_cmd(self, ...)
         return nil, err
     end
 
-    return _read_reply(self, sock)
+    local res, err = _read_reply(self, sock)
+    while _check_msg(self, res) do
+        if rawget(self, "_buffered_msg") == nil then
+            self._buffered_msg = new_tab(1, 0)
+        end
+
+        tab_insert(self._buffered_msg, res)
+        res, err = _read_reply(self, sock)
+    end
+
+    return res, err
 end
 
 
@@ -339,6 +357,8 @@ local function _check_subscribed(self, res)
        and res[3] == 0
    then
         self._subscribed = false
+        -- FIXME: support multiple subscriptions in the next PR
+        self._buffered_msg = nil
     end
 end
 
@@ -353,6 +373,18 @@ function _M.read_reply(self)
         return nil, "not subscribed"
     end
 
+    local buffered_msg = rawget(self, "_buffered_msg")
+    if buffered_msg then
+        local msg = buffered_msg[1]
+        tab_remove(buffered_msg, 1)
+
+        if #buffered_msg == 0 then
+            self._buffered_msg = nil
+        end
+
+        return msg
+    end
+
     local res, err = _read_reply(self, sock)
     _check_subscribed(self, res)
 

t/pubsub.t

@@ -556,3 +556,100 @@ GET /t
 
 --- no_error_log
 [error]
+
+
+
+=== TEST 8: mix read_reply and other commands
+--- http_config eval: $::HttpConfig
+--- config
+    location /t {
+        lua_socket_log_errors off;
+        content_by_lua '
+            local cjson = require "cjson"
+            local redis = require "resty.redis"
+
+            local red = redis:new()
+            local red2 = redis:new()
+
+            red:set_timeout(1000) -- 1 sec
+            red2:set_timeout(1000) -- 1 sec
+
+            local ok, err = red:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT)
+            if not ok then
+                ngx.say("1: failed to connect: ", err)
+                return
+            end
+
+            ok, err = red2:connect("127.0.0.1", $TEST_NGINX_REDIS_PORT)
+            if not ok then
+                ngx.say("2: failed to connect: ", err)
+                return
+            end
+
+            res, err = red:subscribe("dog")
+            if not res then
+                ngx.say("1: failed to subscribe: ", err)
+                return
+            end
+
+            res, err = red2:publish("dog", "Hello")
+            if not res then
+                ngx.say("2: failed to publish: ", err)
+                return
+            end
+
+            res, err = red:ping()
+            if not res then
+                ngx.say("1: failed to subscribe: ", err)
+                return
+            end
+
+            res, err = red2:publish("dog", "World")
+            if not res then
+                ngx.say("2: failed to publish: ", err)
+                return
+            end
+
+            res, err = red:read_reply()
+            if not res then
+                ngx.say("1: failed to read reply: ", err)
+            else
+                ngx.say("1: receive: ", cjson.encode(res))
+            end
+
+            res, err = red:read_reply()
+            if not res then
+                ngx.say("1: failed to read reply: ", err)
+            else
+                ngx.say("1: receive: ", cjson.encode(res))
+            end
+
+            res, err = red:unsubscribe()
+            if not res then
+                ngx.say("1: failed to unscribe: ", err)
+            else
+                ngx.say("1: unsubscribe: ", cjson.encode(res))
+            end
+
+            red:set_timeout(1) -- 1s
+            res, err = red:read_reply()
+            if not res then
+                ngx.say("1: failed to read reply: ", err)
+            else
+                ngx.say("1: receive: ", cjson.encode(res))
+            end
+
+            red:close()
+            red2:close()
+        ';
+    }
+--- request
+GET /t
+--- response_body_like chop
+1: receive: \["message","dog","Hello"\]
+1: receive: \["message","dog","World"\]
+1: unsubscribe: \["unsubscribe","dog",0\]
+1: failed to read reply: not subscribed$
+
+--- no_error_log
+[error]