feature: watch cancel (#73)

* [WIP] feature: implement operation 'compact'

* [WIP] feat: watch cancel

* [WIP] feat: watch cancel

* cancel watch by close http connection

* fix lint

* retest

* retest

* return consistency

* show error clearly

Author: Yiyiyimu

lib/resty/etcd/v3.lua

@@ -513,8 +513,9 @@ local function request_chunk(self, method, host, port, path, opts, timeout)
         return nil, "failed to watch data, response code: " .. res.status
     end
 
-    return function()
-        body, err = res.body_reader()
+    local function read_watch()
+        local body, err = res.body_reader()
+
         if not body then
             return nil, err
         end
@@ -541,6 +542,12 @@ local function request_chunk(self, method, host, port, path, opts, timeout)
 
         return body
     end
+
+    if opts.need_cancel == true then
+        return read_watch, nil, http_cli
+    else
+        return read_watch
+    end
 end
 
 
@@ -602,6 +609,11 @@ local function watch(self, key, attr)
         filters = attr.filters and attr.filters or 0
     end
 
+    local need_cancel
+    if attr.need_cancel then
+        need_cancel = attr.need_cancel and true or false
+    end
+
     local opts = {
         body = {
             create_request = {
@@ -614,19 +626,23 @@ local function watch(self, key, attr)
                 fragment        = fragment,
                 filters         = filters,
             }
-        }
+        },
+        need_cancel = need_cancel,
     }
 
     local endpoint = choose_endpoint(self)
-    local callback_fun, err = request_chunk(self, 'POST',
+
+    local callback_fun, err, http_cli = request_chunk(self, 'POST',
                                 endpoint.host,
                                 endpoint.port,
                                 endpoint.api_prefix .. '/watch', opts,
                                 attr.timeout or self.timeout)
     if not callback_fun then
         return nil, err
     end
-
+    if opts.need_cancel == true then
+        return callback_fun, nil, http_cli
+    end
     return callback_fun
 end
 
@@ -658,10 +674,16 @@ function _M.watch(self, key, opts)
     attr.prev_kv  = opts and opts.prev_kv
     attr.watch_id = opts and opts.watch_id
     attr.fragment = opts and opts.fragment
+    attr.need_cancel = opts and opts.need_cancel
 
     return watch(self, key, attr)
 end
 
+function _M.watchcancel(self, http_cli)
+    local res, err = http_cli:close()
+    return res, err
+end
+
 function _M.readdir(self, key, opts)
 
     clear_tab(attr)
@@ -693,6 +715,7 @@ function _M.watchdir(self, key, opts)
     attr.prev_kv  = opts and opts.prev_kv
     attr.watch_id = opts and opts.watch_id
     attr.fragment = opts and opts.fragment
+    attr.need_cancel = opts and opts.need_cancel
 
     return watch(self, key, attr)
 end

t/v3/key.t

@@ -163,7 +163,66 @@ timeout/
 
 
 
-=== TEST 4: watchdir(key)
+=== TEST 4: watch and watchcancel(key)
+--- http_config eval: $::HttpConfig
+--- config
+    location /t {
+        content_by_lua_block {
+            local etcd, err = require("resty.etcd").new({protocol = "v3"})
+            check_res(etcd, err)
+
+            local res, err = etcd:set("/test", "abc")
+            check_res(res, err)
+
+            ngx.timer.at(0.1, function ()
+                etcd:set("/test", "bcd3")
+            end)
+
+            ngx.timer.at(0.2, function ()
+                etcd:set("/test", "bcd4")
+            end)
+
+            local cur_time = ngx.now()
+            local body_chunk_fun, err, http_cli = etcd:watch("/test", {timeout = 0.5, need_cancel = true})
+
+            if type(http_cli) ~= "table" then
+                ngx.say("need_cancel failed")
+            end
+
+            if not body_chunk_fun then
+                ngx.say("failed to watch: ", err)
+            end
+
+            local chunk, err = body_chunk_fun()
+            ngx.say("created: ", chunk.result.created)
+            local chunk, err = body_chunk_fun()
+            ngx.say("value: ", chunk.result.events[1].kv.value)
+
+            local res, err = etcd:watchcancel(http_cli)
+            if not res then
+                ngx.say("failed to cancel: ", err)
+            end
+
+            local chunk, err = body_chunk_fun()
+            ngx.say(err)
+
+            ngx.say("ok")
+        }
+    }
+--- request
+GET /t
+--- no_error_log
+[error]
+--- response_body
+created: true
+value: bcd3
+closed
+ok
+--- timeout: 5
+
+
+
+=== TEST 5: watchdir(key)
 --- http_config eval: $::HttpConfig
 --- config
     location /t {
@@ -289,7 +348,7 @@ timeout/
 
 
 
-=== TEST 5: setx(key, val) failed
+=== TEST 6: setx(key, val) failed
 --- http_config eval: $::HttpConfig
 --- config
     location /t {
@@ -310,7 +369,7 @@ GET /t
 
 
 
-=== TEST 6: setx(key, val) success
+=== TEST 7: setx(key, val) success
 --- http_config eval: $::HttpConfig
 --- config
     location /t {
@@ -336,7 +395,7 @@ checked val as expect: abd
 
 
 
-=== TEST 7: setnx(key, val)
+=== TEST 8: setnx(key, val)
 --- http_config eval: $::HttpConfig
 --- config
     location /t {

t/v3/lease3.4.t

@@ -65,6 +65,7 @@ __DATA__
             local data, err = etcd:leases()
             if data.body.leases[1].ID ~= res.body.ID then
                 ngx.say("leases not working")
+                ngx.say("result: ", require("cjson").encode(data.body))
             end
 
             local data, err = etcd:set("/test", "abc", {prev_kv = true, lease = res.body.ID})