You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by di...@apache.org on 2019/10/22 08:01:00 UTC
[rocketmq-client-go] branch master updated: fix(transaction):can
not do local execute and check (#262)
This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-client-go.git
The following commit(s) were added to refs/heads/master by this push:
new 6bb5243 fix(transaction):can not do local execute and check (#262)
6bb5243 is described below
commit 6bb52431a5eede4b3781d13a90120dea46779669
Author: dinglei <li...@163.com>
AuthorDate: Tue Oct 22 16:00:53 2019 +0800
fix(transaction):can not do local execute and check (#262)
---
core/api.go | 3 +++
core/transaction_funcs.go | 12 ++++++------
demos/transaction_producer.go | 4 ++--
3 files changed, 11 insertions(+), 8 deletions(-)
diff --git a/core/api.go b/core/api.go
index f141f3f..a8941b6 100644
--- a/core/api.go
+++ b/core/api.go
@@ -56,6 +56,7 @@ type ProducerModel int
const (
CommonProducer = ProducerModel(1)
OrderlyProducer = ProducerModel(2)
+ TransProducer = ProducerModel(3)
)
func (mode ProducerModel) String() string {
@@ -64,6 +65,8 @@ func (mode ProducerModel) String() string {
return "CommonProducer"
case OrderlyProducer:
return "OrderlyProducer"
+ case TransProducer:
+ return "TransProducer"
default:
return "Unknown"
}
diff --git a/core/transaction_funcs.go b/core/transaction_funcs.go
index a41dd4c..ddab6aa 100644
--- a/core/transaction_funcs.go
+++ b/core/transaction_funcs.go
@@ -39,10 +39,10 @@ func localTransactionExecutorCallback(cproducer *C.CProducer, msg *C.CMessage, a
message := cMsgToGo(msg)
listenerWrap, exist := producer.(*defaultTransactionProducer).listenerFuncsMap.Load(cproducer)
if !exist {
- status := listenerWrap.(TransactionLocalListener).Execute(message, arg)
- return C.int(status)
+ return C.int(UnknownTransaction)
}
- return C.int(UnknownTransaction)
+ status := listenerWrap.(TransactionLocalListener).Execute(message, arg)
+ return C.int(status)
}
//export localTransactionCheckerCallback
@@ -55,8 +55,8 @@ func localTransactionCheckerCallback(cproducer *C.CProducer, msg *C.CMessageExt,
message := cmsgExtToGo(msg)
listener, exist := producer.(*defaultTransactionProducer).listenerFuncsMap.Load(cproducer)
if !exist {
- status := listener.(TransactionLocalListener).Check(message, arg)
- return C.int(status)
+ return C.int(UnknownTransaction)
}
- return C.int(UnknownTransaction)
+ status := listener.(TransactionLocalListener).Check(message, arg)
+ return C.int(status)
}
diff --git a/demos/transaction_producer.go b/demos/transaction_producer.go
index a4b64b2..8cf4112 100644
--- a/demos/transaction_producer.go
+++ b/demos/transaction_producer.go
@@ -35,8 +35,8 @@ func main4() {
Channel: "ALIYUN/OtherChannel",
},
},
- //Set to Common Producer as default.
- ProducerModel: rocketmq.CommonProducer,
+ //Set to Trans Producer as default.
+ ProducerModel: rocketmq.TransProducer,
}
sendTransactionMessage(pConfig)
}