change api

README.md

@@ -18,8 +18,9 @@ Table of Contents
             * [setUseTLS](#setUseTLS)
             * [registerSendMessageHook](#registerSendMessageHook)
             * [registerEndTransactionHook](#registerEndTransactionHook)
+            * [send](#send)
+            * [sendMessageInTransaction](#sendMessageInTransaction)
             * [start](#start)
-            * [produce](#produce)
             * [stop](#stop)
     * [resty.rocketmq.admin](#restyrocketmqadmin)
       * [Methods](#methods)
@@ -78,7 +79,7 @@ Synopsis
                 -- use tls mode
                 p:setUseTLS(true)
 
-                local res, err = p:produce("TopicTest", message)
+                local res, err = p:send("TopicTest", message)
                 if not res then
                     ngx.say("send err:", err)
                     return
@@ -176,18 +177,29 @@ there is an acl hook provided, usage is:
   - transactionState
   - fromTransactionCheck
 
-#### start
-`syntax: p:start()`
-
-note that if you don't call p:start() before sending messages, messages will be sent successfully, but the trace is not send.
 
-
-#### produce
-`syntax: res, err = p:produce(topic, message, tags, keys, waitStoreMsgOk)`
+#### send
+`syntax: res, err = p:send(topic, message, tags, keys, waitStoreMsgOk)`
 
   In case of success, returns the a table of results.
   In case of errors, returns `nil` with a string describing the error.
 
+#### setTransactionListener
+`syntax: res, err = p:setTransactionListener(transactionListener)`
+
+`transactionListener` is a table that contains two functions as follows:
+
+- `executeLocalTransaction(self, msg, arg)`
+- `checkLocalTransaction(self, msg)`
+
+#### sendMessageInTransaction
+`syntax: res, err = p:sendMessageInTransaction(topic, arg, message, tags, keys, waitStoreMsgOk)`
+
+#### start
+`syntax: p:start()`
+
+note that if you don't call p:start() before sending messages, messages will be sent successfully, but the trace is not send.
+
 #### stop
 `syntax: p:stop()`
 

lib/resty/rocketmq/producer.lua

@@ -3,6 +3,7 @@ local client = require("resty.rocketmq.client")
 local utils = require("resty.rocketmq.utils")
 local trace = require("resty.rocketmq.trace")
 local RESPONSE_CODE = core.RESPONSE_CODE
+local cjson_safe = require("cjson.safe")
 
 ---@class producer
 local _M = {}
@@ -69,16 +70,7 @@ local function sendHeartbeatToAllBroker(self)
     local heartbeatData = {
         clientID = '' .. ngx.worker.pid(),
         producerDataSet = { { groupName = self.groupName } },
-        consumerDataSet = {
-            {
-                groupName = self.groupName,
-                consumerType = self.groupName,
-                messageModel = self.groupName,
-                consumerFromWhere = self.groupName,
-                subscriptionDataSet = self.groupName,
-                unitMode = self.groupName,
-            }
-        }
+        consumerDataSet = setmetatable({}, cjson_safe.empty_array_mt)
     }
     for brokerName, brokers in pairs(self.client.brokerAddrTable) do
         local addr = brokers[0]
@@ -186,7 +178,7 @@ local function produce(self, msg)
     return h
 end
 
-function _M:produce(topic, message, tags, keys, waitStoreMsgOk)
+function _M:send(topic, message, tags, keys, waitStoreMsgOk)
     return produce(self, {
         producerGroup = self.groupName,
         topic = topic,
@@ -209,11 +201,18 @@ function _M:produce(topic, message, tags, keys, waitStoreMsgOk)
     })
 end
 
+function _M:setTransactionListener(transactionListener)
+    if type(transactionListener.executeLocalTransaction) ~= 'function' then
+        return nil, 'invalid callback'
+    end
+    self.transactionListener = transactionListener
+end
+
 
 -- todo add check callback
-function _M:transactionProduce(topic, execute, arg, message, tags, keys, waitStoreMsgOk)
-    if type(execute) ~= 'function' then
-        return nil, 'invalid callback'
+function _M:sendMessageInTransaction(topic, arg, message, tags, keys, waitStoreMsgOk)
+    if not self.transactionListener then
+        return nil, "TransactionListener is null"
     end
     local msg = {
         producerGroup = self.groupName,
@@ -245,7 +244,7 @@ function _M:transactionProduce(topic, execute, arg, message, tags, keys, waitSto
     if h.code == RESPONSE_CODE.SUCCESS then
         msg.properties.__transationId__ = h.transationId
         msg.transationId = msg.properties.UNIQ_KEY
-        localTransactionState = execute(msg, arg)
+        localTransactionState = self.transactionListener:executeLocalTransaction(msg, arg)
     else
         localTransactionState = core.TRANSACTION_ROLLBACK_TYPE
     end

lib/resty/rocketmq/trace.lua

@@ -217,7 +217,7 @@ local function sendTraceDataByMQ(self, keySet, data, topic)
     for k in pairs(keySet) do
         table.insert(keys, k)
     end
-    local res, err = self.producer:produce(core.RMQ_SYS_TRACE_TOPIC, data, "", table.concat(keys, ' '))
+    local res, err = self.producer:send(core.RMQ_SYS_TRACE_TOPIC, data, "", table.concat(keys, ' '))
     if err then
         ngx.log(ngx.WARN, 'send msg trace fail, ', err)
     end

t/producer.t

@@ -65,7 +65,7 @@ ok
                 return
             end
 
-            local res, err = p:produce("TopicTest", message)
+            local res, err = p:send("TopicTest", message)
             if not res then
                 ngx.say("send err:", err)
                 return

t/trace.t

@@ -42,7 +42,7 @@ __DATA__
                 return
             end
             p:start()
-            local res, err = p:produce("TopicTest", "halo world", "tags", "keys")
+            local res, err = p:send("TopicTest", "halo world", "tags", "keys")
             if not res then
                 ngx.say("send err:", err)
                 return