add setTimeout for producer and admin

yuz10 · Nov 30, 2021 · 7c0e598 · 7c0e598
1 parent 49a750b
commit 7c0e598
Show file tree

Hide file tree

Showing 6 changed files with 37 additions and 6 deletions.
diff --git a/README.md b/README.md
@@ -16,6 +16,7 @@ Table of Contents
             * [new](#new)
             * [addRPCHook](#addRPCHook)
             * [setUseTLS](#setUseTLS)
+            * [setTimeout](#setTimeout)
             * [registerSendMessageHook](#registerSendMessageHook)
             * [registerEndTransactionHook](#registerEndTransactionHook)
             * [send](#send)
@@ -27,6 +28,7 @@ Table of Contents
           * [new](#new-1)
           * [addRPCHook](#addRPCHook-1)
           * [setUseTLS](#setUseTLS-1)
+          * [setTimeout](#setTimeout-1)
           * [createTopic](#createTopic)
           * [createTopicForBroker](#createTopicForBroker)
           * [searchOffset](#searchOffset)
@@ -139,6 +141,12 @@ there is an acl hook provided, usage is:
 
 `useTLS` is a boolean
 
+#### setTimeout
+
+`syntax: p:setTimeout(timeout)`
+
+`timeout` is in milliseconds, default 3000
+
 #### registerSendMessageHook
 
 `syntax: p:registerSendMessageHook(hook)`
@@ -249,6 +257,12 @@ there is an acl hook provided, usage is:
 
 `useTLS` is a boolean
 
+#### setTimeout
+
+`syntax: p:setTimeout(timeout)`
+
+`timeout` is in milliseconds, default 3000
+
 #### createTopic
 `syntax: res, err = adm:createTopic(defaultTopic, newTopic, queueNum, topicSysFlag)`
 

diff --git a/lib/resty/rocketmq/admin.lua b/lib/resty/rocketmq/admin.lua
@@ -30,6 +30,10 @@ function _M.setUseTLS(self, useTLS)
     self.client:setUseTLS(useTLS)
 end
 
+function _M.setTimeout(self, timeout)
+    self.client:setTimeout(timeout)
+end
+
 function _M.createTopic(self, defaultTopic, newTopic, queueNum, topicSysFlag)
     if not core.checkTopic(newTopic) then
         return nil, ('topic %s invalid format'):format(newTopic)

diff --git a/lib/resty/rocketmq/client.lua b/lib/resty/rocketmq/client.lua
@@ -34,6 +34,7 @@ function _M.new(nameservers)
         current_nameserver = 1,
         RPCHook = {},
         useTLS = false,
+        timeout = 3000,
 
         topicPublishInfoTable = {},
         brokerAddrTable = {},
@@ -52,8 +53,12 @@ function _M.setUseTLS(self, useTLS)
     self.useTLS = useTLS
 end
 
+function _M.setTimeout(self, timeout)
+    self.timeout = timeout
+end
+
 function _M:request(code, addr, header, body, oneway)
-    return core.request(code, addr, header, body, oneway, self.RPCHook, self.useTLS)
+    return core.request(code, addr, header, body, oneway, self.RPCHook, self.useTLS, self.timeout)
 end
 
 function _M:chooseNameserver()

diff --git a/lib/resty/rocketmq/core.lua b/lib/resty/rocketmq/core.lua
@@ -290,8 +290,9 @@ local function getLong(buffer, offset)
     return lshift(res1, 32) + res2, offset
 end
 
-local function doReqeust(ip, port, send, oneway, useTLS)
+local function doReqeust(ip, port, send, oneway, useTLS, timeout)
     local sock = ngx_socket_tcp()
+    sock:settimeout(timeout)
     local res, err = sock:connect(ip, port)
     if not res then
         return nil, nil, ('connect %s:%s fail:%s'):format(ip, port, err)
@@ -328,7 +329,7 @@ local function doReqeust(ip, port, send, oneway, useTLS)
 end
 _M.doReqeust = doReqeust
 
-local function request(code, addr, header, body, oneway, RPCHook, useTLS)
+local function request(code, addr, header, body, oneway, RPCHook, useTLS, timeout)
     if RPCHook then
         for _, hook in ipairs(RPCHook) do
             hook:doBeforeRequest(addr, header, body)
@@ -337,7 +338,7 @@ local function request(code, addr, header, body, oneway, RPCHook, useTLS)
     ngx.log(ngx.DEBUG, ('\27[33msend: %s %s\27[0m %s %s'):format(addr, REQUEST_CODE_NAME[code] or code, cjson_safe.encode(header), body))
     local send = encode(code, header, body, oneway)
     local ip, port = unpack(split(addr, ':'))
-    local respHeader, respBody, err = doReqeust(ip, port, send, oneway, useTLS)
+    local respHeader, respBody, err = doReqeust(ip, port, send, oneway, useTLS, timeout)
     if err then
         return nil, nil, err
     end

diff --git a/lib/resty/rocketmq/producer.lua b/lib/resty/rocketmq/producer.lua
@@ -43,6 +43,13 @@ function _M.setUseTLS(self, useTLS)
     end
 end
 
+function _M.setTimeout(self, timeout)
+    self.client:setTimeout(timeout)
+    if self.traceDispatcher then
+        self.traceDispatcher.producer:setTimeout(timeout)
+    end
+end
+
 function _M.registerSendMessageHook(self, hook)
     if type(hook) == 'table' and type(hook.sendMessageBefore) == 'function' and type(hook.sendMessageAfter) == 'function' then
         table.insert(self.sendMessageHookList, hook)

diff --git a/lua-resty-rocketmq-0.2.0-0.rockspec b/lua-resty-rocketmq-0.2.0-0.rockspec
@@ -1,8 +1,8 @@
 package = "lua-resty-rocketmq"
-version = "0.2.0-0"
+version = "0.2.1-0"
 source = {
    url = "git://github.com/yuz10/lua-resty-rocketmq.git",
-   tag = "0.2.0"
+   tag = "0.2.1"
 }
 description = {
    summary = "Lua RocketMQ client driver for the ngx_lua based on the cosocket API",