This is an automated email from the ASF dual-hosted git repository.
ssshr-66 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new baabdf73 feature: add RocketMQ TCC integration (#1054)
baabdf73 is described below
commit baabdf73cf6a99fffef3a539bcc46e229b513866
Author: CAICAII <[email protected]>
AuthorDate: Sat Jun 6 03:41:11 2026 +0800
feature: add RocketMQ TCC integration (#1054)
* feature: add RocketMQ TCC integration
- Implement TCCRocketMQAction with Prepare/Commit/Rollback
- Add SeataMQProducer wrapper for transaction message sending
- Add SeataTransactionListener for broker check-back
- Add producer factory with singleton pattern
- Add XID/BranchId propagation via message properties
* fix: address PR review comments
- Add safe type assertion in CheckLocalTransaction
- Replace sync.Once with mutex for retry support
- Remove duplicate ParseTCCResource call
- Add separate normalProducer for non-transactional messages
- Document Commit/Rollback delegation model
- Add explicit handling for intermediate global statuses
- Remove unused ActionContextKeySendResult constant
* fix: resolve MessageFuture race condition causing sync timeout
* test: harden rocketmq tcc integration
* fix: treat finished global status as unknown
* refactor: improve rocketmq integration readability
* test: strengthen rocketmq producer coverage
---------
Co-authored-by: 福來｜Fulai <[email protected]>
Co-authored-by: ThunGuo <[email protected]>
Co-authored-by: flypiggyNo3 <[email protected]>
Co-authored-by: ssshr-66 <[email protected]>
---
go.mod | 7 +-
go.sum | 41 +++++
pkg/constant/context.go | 4 +
pkg/integration/rocketmq/config.go | 57 ++++++
.../rocketmq/constants.go} | 27 +--
pkg/integration/rocketmq/producer_factory.go | 80 +++++++++
pkg/integration/rocketmq/producer_factory_test.go | 170 ++++++++++++++++++
pkg/integration/rocketmq/seata_producer.go | 193 +++++++++++++++++++++
pkg/integration/rocketmq/seata_producer_test.go | 177 +++++++++++++++++++
pkg/integration/rocketmq/tcc_rocketmq_action.go | 104 +++++++++++
.../rocketmq/tcc_rocketmq_action_test.go | 93 ++++++++++
pkg/integration/rocketmq/test_helpers_test.go | 96 ++++++++++
pkg/integration/rocketmq/transaction_listener.go | 116 +++++++++++++
.../rocketmq/transaction_listener_test.go | 157 +++++++++++++++++
pkg/protocol/message/other_message.go | 2 +-
pkg/protocol/message/other_message_test.go | 4 +-
pkg/remoting/getty/getty_remoting.go | 6 +-
pkg/remoting/getty/getty_remoting_test.go | 46 +++++
18 files changed, 1358 insertions(+), 22 deletions(-)
diff --git a/go.mod b/go.mod
index 39b5adc9..decb9e08 100644
--- a/go.mod
+++ b/go.mod
@@ -6,6 +6,7 @@ require (
dubbo.apache.org/dubbo-go/v3 v3.0.4
github.com/DATA-DOG/go-sqlmock v1.5.0
github.com/apache/dubbo-getty v1.5.0
+ github.com/apache/rocketmq-client-go/v2 v2.1.2
github.com/arana-db/parser v0.2.17
github.com/bluele/gcache v0.0.2
github.com/dsnet/compress v0.0.1
@@ -84,6 +85,7 @@ require (
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd //
indirect
github.com/modern-go/reflect2 v1.0.2 // indirect
github.com/mschoch/smat v0.2.0 // indirect
+ github.com/patrickmn/go-cache v2.1.0+incompatible // indirect
github.com/pelletier/go-toml/v2 v2.0.8 // indirect
github.com/pingcap/errors v0.11.5-0.20210425183316-da1aaba5fb63 //
indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
@@ -92,8 +94,12 @@ require (
github.com/prometheus/procfs v0.7.3 // indirect
github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b //
indirect
github.com/shirou/gopsutil/v3 v3.22.2 // indirect
+ github.com/sirupsen/logrus v1.7.0 // indirect
github.com/stoewer/go-strcase v1.2.0 // indirect
github.com/stretchr/objx v0.5.0 // indirect
+ github.com/tidwall/gjson v1.13.0 // indirect
+ github.com/tidwall/match v1.1.1 // indirect
+ github.com/tidwall/pretty v1.2.0 // indirect
github.com/tklauser/go-sysconf v0.3.10 // indirect
github.com/tklauser/numcpus v0.4.0 // indirect
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
@@ -111,7 +117,6 @@ require (
)
require (
- github.com/BurntSushi/toml v1.1.0 // indirect
github.com/go-playground/validator/v10 v10.14.0 // indirect
github.com/klauspost/compress v1.15.11
github.com/mattn/go-colorable v0.1.8 // indirect
diff --git a/go.sum b/go.sum
index e1eb50e9..141da943 100644
--- a/go.sum
+++ b/go.sum
@@ -74,6 +74,8 @@ github.com/apache/dubbo-go-hessian2 v1.9.1/go.mod
h1:xQUjE7F8PX49nm80kChFvepA/Av
github.com/apache/dubbo-go-hessian2 v1.9.3/go.mod
h1:xQUjE7F8PX49nm80kChFvepA/AvqAZ0oh/UaB6+6pBE=
github.com/apache/dubbo-go-hessian2 v1.11.4
h1:u8spEogpaZ82/k3bmv1PLYnha/wiwsIme15FuZZgkiI=
github.com/apache/dubbo-go-hessian2 v1.11.4/go.mod
h1:QP9Tc0w/B/mDopjusebo/c7GgEfl6Lz8jeuFg8JA6yw=
+github.com/apache/rocketmq-client-go/v2 v2.1.2
h1:yt73olKe5N6894Dbm+ojRf/JPiP0cxfDNNffKwhpJVg=
+github.com/apache/rocketmq-client-go/v2 v2.1.2/go.mod
h1:6I6vgxHR3hzrvn+6n/4mrhS+UTulzK/X9LB2Vk1U5gE=
github.com/apache/thrift v0.12.0/go.mod
h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
github.com/apache/thrift v0.13.0/go.mod
h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
github.com/arana-db/parser v0.2.17
h1:4wNfSgza2N3pjpwR5jmWLvu4L6Sme6EtoLuZOgwWlsU=
@@ -170,6 +172,7 @@ github.com/davecgh/go-spew v1.1.0/go.mod
h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
github.com/davecgh/go-spew v1.1.1
h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod
h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod
h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod
h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod
h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
github.com/dsnet/compress v0.0.1
h1:PlZu0n3Tuv04TzpfPbrnI0HW/YwodEXDS+oPKahKF0Q=
github.com/dsnet/compress v0.0.1/go.mod
h1:Aw8dCMJ7RioblQeTqt88akK31OvO8Dhf5JflhBbQEHo=
@@ -189,6 +192,7 @@ github.com/eapache/go-xerial-snappy
v0.0.0-20180814174437-776d5712da21/go.mod h1
github.com/eapache/queue v1.1.0/go.mod
h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
github.com/edsrzf/mmap-go v1.0.0/go.mod
h1:YO35OhQPt3KJa3ryjFM5Bs14WD66h8eGKpfaBNrHW5M=
github.com/emicklei/go-restful/v3 v3.10.1/go.mod
h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
+github.com/emirpasic/gods v1.12.0/go.mod
h1:YfzfFFoVP/catgzJb4IKIqXjX78Ha8FMSDh3ymbK86o=
github.com/envoyproxy/go-control-plane v0.6.9/go.mod
h1:SBwIajubJHhxtWwsL9s8ss4safvEdbitLhGGK48rN6g=
github.com/envoyproxy/go-control-plane v0.9.0/go.mod
h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
github.com/envoyproxy/go-control-plane
v0.9.1-0.20191026205805-5f8ba28d4473/go.mod
h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
@@ -255,11 +259,13 @@ github.com/go-playground/universal-translator
v0.18.1/go.mod h1:xekY+UJKNuX9WP91
github.com/go-playground/validator/v10 v10.11.0/go.mod
h1:i+3WkQ1FvaUjjxh1kSvIA4dMGDBiPU55YFDl0WbKdWU=
github.com/go-playground/validator/v10 v10.14.0
h1:vgvQWe3XCz3gIeFDm/HnTIbj6UGmg/+t63MyGU2n5js=
github.com/go-playground/validator/v10 v10.14.0/go.mod
h1:9iXMNT7sEkjXb0I+enO7QXmzG6QCsPWY4zveKFVRSyU=
+github.com/go-redis/redis/v8 v8.11.5/go.mod
h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
github.com/go-resty/resty/v2 v2.7.0/go.mod
h1:9PWDzw47qPphMRFfhsyk0NnSgvluHcljSMVIq3w7q0I=
github.com/go-sql-driver/mysql v1.4.0/go.mod
h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
github.com/go-sql-driver/mysql v1.6.0
h1:BCTh4TKNUYmOmMUcQ3IipzF5prigylS7XXjEkfCHuOE=
github.com/go-sql-driver/mysql v1.6.0/go.mod
h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
github.com/go-stack/stack v1.8.0/go.mod
h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
+github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod
h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE=
github.com/go-test/deep v1.0.2-0.20181118220953-042da051cf31/go.mod
h1:wGDj63lr65AM2AQyKZd/NYHGb0R+1RLqB8NKt3aSFNA=
github.com/go-test/deep v1.0.2/go.mod
h1:wGDj63lr65AM2AQyKZd/NYHGb0R+1RLqB8NKt3aSFNA=
github.com/goccy/go-json v0.10.2
h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
@@ -343,6 +349,7 @@ github.com/google/pprof
v0.0.0-20200212024743-f11f1df84d12/go.mod h1:ZgVRPoUq/hf
github.com/google/pprof v0.0.0-20200229191704-1ebb73c60ed3/go.mod
h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
github.com/google/pprof v0.0.0-20200430221834-fc25d7d30c6d/go.mod
h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
github.com/google/pprof v0.0.0-20200708004538-1a94d8640e99/go.mod
h1:ZgVRPoUq/hfqzAqh7sHMqb3I9Rq5C59dIz2SbBwJ4eM=
+github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38/go.mod
h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
github.com/google/renameio v0.1.0/go.mod
h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
github.com/google/uuid v1.0.0/go.mod
h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.1.1/go.mod
h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
@@ -352,6 +359,7 @@ github.com/google/uuid v1.3.0/go.mod
h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
github.com/googleapis/gax-go/v2 v2.0.4/go.mod
h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
github.com/googleapis/gax-go/v2 v2.0.5/go.mod
h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod
h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
+github.com/gopherjs/gopherjs v0.0.0-20190910122728-9d188e94fb99
h1:twflg0XRTjwKpxb/jFExr4HGq6on2dEOmnL6FV+fgPw=
github.com/gopherjs/gopherjs v0.0.0-20190910122728-9d188e94fb99/go.mod
h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
github.com/gorilla/context v1.1.1/go.mod
h1:kBGZzfjB9CEq2AlWe17Uuf7NDRt0dE0s8S51q0aT7Yg=
github.com/gorilla/mux v1.6.2/go.mod
h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
@@ -436,6 +444,7 @@ github.com/hjson/hjson-go/v4 v4.0.0/go.mod
h1:KaYt3bTw3zhBjYqnXkYywcYctk0A2nxeEF
github.com/hpcloud/tail v1.0.0/go.mod
h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
github.com/hudl/fargo v1.3.0/go.mod
h1:y3CKSmjA+wD2gak7sUSXTAoopbhU08POFhmITJgmKTg=
github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod
h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
+github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod
h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
github.com/inconshreveable/mousetrap v1.0.0/go.mod
h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
github.com/influxdata/influxdb1-client
v0.0.0-20191209144304-8bf82d3c094d/go.mod
h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
github.com/jackc/pgpassfile v1.0.0
h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
@@ -467,6 +476,7 @@ github.com/json-iterator/go v1.1.12
h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr
github.com/json-iterator/go v1.1.12/go.mod
h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod
h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
github.com/jstemmer/go-junit-report v0.9.1/go.mod
h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
+github.com/jtolds/gls v4.20.0+incompatible
h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
github.com/jtolds/gls v4.20.0+incompatible/go.mod
h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
github.com/julienschmidt/httprouter v1.2.0/go.mod
h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
github.com/julienschmidt/httprouter v1.3.0/go.mod
h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
@@ -580,6 +590,8 @@ github.com/nats-io/nkeys v0.1.3/go.mod
h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxzi
github.com/nats-io/nuid v1.0.1/go.mod
h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod
h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
github.com/npillmayer/nestext v0.1.3/go.mod
h1:h2lrijH8jpicr25dFY+oAJLyzlya6jhnuG+zWp9L0Uk=
+github.com/nxadm/tail v1.4.4/go.mod
h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
+github.com/nxadm/tail v1.4.8/go.mod
h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
github.com/oklog/oklog v0.3.2/go.mod
h1:FCV+B7mhrz4o+ueLpx+KqkyXRGMWOYEvfiXtdGtbWGs=
github.com/oklog/run v1.0.0/go.mod
h1:dlhp/R75TPv97u0XWUtDeV/lRKWPKSdTuV0TZvrmrQA=
github.com/oklog/ulid v1.3.1/go.mod
h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
@@ -587,7 +599,15 @@ github.com/olekukonko/tablewriter
v0.0.0-20170122224234-a0225b3f23b5/go.mod h1:v
github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852/go.mod
h1:eqOVx5Vwu4gd2mmMZvVZsgIqNSaW3xxRThUJ0k/TPk4=
github.com/onsi/ginkgo v1.6.0/go.mod
h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.7.0/go.mod
h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.12.1/go.mod
h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
+github.com/onsi/ginkgo v1.16.4/go.mod
h1:dX+/inL/fNMqNlz0e9LfyB9TswhZpCVdJM/Z6Vvnwo0=
+github.com/onsi/ginkgo v1.16.5/go.mod
h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
+github.com/onsi/ginkgo/v2 v2.0.0/go.mod
h1:vw5CSIxN1JObi/U8gcbwft7ZxR2dgaR70JSE3/PpL4c=
github.com/onsi/gomega v1.4.3/go.mod
h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
+github.com/onsi/gomega v1.7.1/go.mod
h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
+github.com/onsi/gomega v1.10.1/go.mod
h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
+github.com/onsi/gomega v1.17.0/go.mod
h1:HnhC7FXeEQY45zxNK3PPoIUhzk/80Xly9PcubAlGdZY=
+github.com/onsi/gomega v1.18.1/go.mod
h1:0q+aL8jAiMXy9hbwj2mr5GziHiwhAIQpFmmtT5hitRs=
github.com/op/go-logging v0.0.0-20160315200505-970db520ece7/go.mod
h1:HzydrMdWErDVzsI23lYNej1Htcns9BCg93Dk0bBINWk=
github.com/opentracing-contrib/go-observer
v0.0.0-20170622124052-a52f23424492/go.mod
h1:Ngi6UdF0k5OKD5t5wlmGhe/EDKPoUM3BXZSSfIuJbis=
github.com/opentracing/basictracer-go v1.0.0/go.mod
h1:QfBfYuafItcjQuMwinw9GhYKwFXS9KnPs5lxoYwgW74=
@@ -601,6 +621,8 @@ github.com/openzipkin/zipkin-go v0.2.2/go.mod
h1:NaW6tEwdmWMaCDZzg8sh+IBNOxHMPnh
github.com/pact-foundation/pact-go v1.0.4/go.mod
h1:uExwJY4kCzNPcHRj+hCR/HBbOOIwwtUjcrb0b5/5kLM=
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod
h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
github.com/pascaldekloe/goe v0.1.0/go.mod
h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
+github.com/patrickmn/go-cache v2.1.0+incompatible
h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
+github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod
h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
github.com/pborman/uuid v1.2.0/go.mod
h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k=
github.com/pelletier/go-toml v1.2.0/go.mod
h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
github.com/pelletier/go-toml v1.7.0/go.mod
h1:vwGMzjaWMwyfHwgIBhI2YUM4fB6nL6lVAvS1LBMMhTE=
@@ -702,11 +724,15 @@ github.com/shirou/gopsutil/v3 v3.22.2
h1:wCrArWFkHYIdDxx/FSfF5RB4dpJYW6t7rcp3+zL
github.com/shirou/gopsutil/v3 v3.22.2/go.mod
h1:WapW1AOOPlHyXr+yOyw3uYx36enocrtSoSBy0L5vUHY=
github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod
h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
github.com/sirupsen/logrus v1.2.0/go.mod
h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
+github.com/sirupsen/logrus v1.4.0/go.mod
h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
github.com/sirupsen/logrus v1.4.2/go.mod
h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
github.com/sirupsen/logrus v1.6.0/go.mod
h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
+github.com/sirupsen/logrus v1.7.0
h1:ShrD1U9pZB12TX0cVy0DtePoCH97K8EtX+mg7ZARUtM=
github.com/sirupsen/logrus v1.7.0/go.mod
h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
+github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d
h1:zE9ykElWQ6/NYmHa3jpm/yHnI4xSofP+UP6SpjHcSeM=
github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod
h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
github.com/smartystreets/goconvey v0.0.0-20190330032615-68dc04aab96a/go.mod
h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
+github.com/smartystreets/goconvey v1.6.4
h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s=
github.com/smartystreets/goconvey v1.6.4/go.mod
h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
github.com/soheilhy/cmux v0.1.4/go.mod
h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
github.com/soheilhy/cmux v0.1.5-0.20210205191134-5ec6847320e5/go.mod
h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
@@ -745,6 +771,12 @@ github.com/stretchr/testify v1.8.2/go.mod
h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o
github.com/stretchr/testify v1.8.3
h1:RP3t2pwF7cMEbC1dqtB6poj3niw/9gnV4Cjg5oW5gtY=
github.com/stretchr/testify v1.8.3/go.mod
h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
github.com/subosito/gotenv v1.2.0/go.mod
h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
+github.com/tidwall/gjson v1.13.0
h1:3TFY9yxOQShrvmjdM76K+jc66zJeT6D3/VFFYCGQf7M=
+github.com/tidwall/gjson v1.13.0/go.mod
h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
+github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
+github.com/tidwall/match v1.1.1/go.mod
h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
+github.com/tidwall/pretty v1.2.0
h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
+github.com/tidwall/pretty v1.2.0/go.mod
h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
github.com/tklauser/go-sysconf v0.3.6/go.mod
h1:MkWzOF4RMCshBAMXuhXJs64Rte09mITnppBXY/rYEFI=
github.com/tklauser/go-sysconf v0.3.9/go.mod
h1:11DU/5sG7UexIrp/O6g35hrWzu0JxlwQ3LSFUzyeuhs=
github.com/tklauser/go-sysconf v0.3.10
h1:IJ1AZGZRWbY8T5Vfk04D9WOA5WSejdflXxP03OUqALw=
@@ -810,6 +842,7 @@ go.opentelemetry.io/proto/otlp v0.7.0/go.mod
h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqe
go.uber.org/atomic v1.3.2/go.mod
h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
go.uber.org/atomic v1.4.0/go.mod
h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
go.uber.org/atomic v1.5.0/go.mod
h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
+go.uber.org/atomic v1.5.1/go.mod
h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
go.uber.org/atomic v1.6.0/go.mod
h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
go.uber.org/atomic v1.7.0/go.mod
h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE=
@@ -930,6 +963,7 @@ golang.org/x/net v0.0.0-20200421231249-e086a090c8fd/go.mod
h1:qpuaurCH72eLCgpAm/
golang.org/x/net v0.0.0-20200501053045-e0ff5e5a1de5/go.mod
h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
golang.org/x/net v0.0.0-20200506145744-7e3656a0809f/go.mod
h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
golang.org/x/net v0.0.0-20200513185701-a91f0712d120/go.mod
h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod
h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
golang.org/x/net v0.0.0-20200520182314-0ba52f642ac2/go.mod
h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod
h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod
h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
@@ -940,6 +974,7 @@ golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod
h1:sp8m0HH+o8qH0wwXwY
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod
h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod
h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1/go.mod
h1:9tjilg8BloeKEkVJvy7fQ90B1CfIiPueXVOjqfkSzI8=
+golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod
h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod
h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
golang.org/x/net v0.0.0-20211029224645-99673261e6eb/go.mod
h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
golang.org/x/net v0.0.0-20211105192438-b53810dc28af/go.mod
h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
@@ -989,6 +1024,7 @@ golang.org/x/sys v0.0.0-20190606165138-5da285871e9c/go.mod
h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190826190057-c7b8b68b1456/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190904154756-749cb33beabd/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190922100055-0a153f010e69/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190924154521-2837fb4f24fe/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -996,6 +1032,7 @@ golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod
h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191008105621-543471e840be/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191220142924-d4481acd189f/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191228213918-04cbcbbfeed8/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -1023,6 +1060,7 @@ golang.org/x/sys
v0.0.0-20201009025420-dfb3f7c4e634/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20201214210602-f9fddec55a1e/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210303074136-134d130e1a04/go.mod
h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -1037,6 +1075,7 @@ golang.org/x/sys
v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.0.0-20210806184541-e5e7981a1069/go.mod
h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20210816074244-15123e1e1f71/go.mod
h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20211106132015-ebca88c72f68/go.mod
h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod
h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220111092808-5a964db01320/go.mod
h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod
h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod
h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -1120,6 +1159,7 @@ golang.org/x/tools
v0.0.0-20200729194436-6467de6f59a7/go.mod h1:njjCfa9FT2d7l9Bc
golang.org/x/tools v0.0.0-20200804011535-6c149bb5ef0d/go.mod
h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
golang.org/x/tools v0.0.0-20200825202427-b303f430e36d/go.mod
h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
golang.org/x/tools v0.0.0-20201014170642-d1624618ad65/go.mod
h1:z6u4i615ZeAfBE4XtMziQW1fSVJXACjjbWkB/mvPzlU=
+golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod
h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod
h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
golang.org/x/tools v0.1.1/go.mod
h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
golang.org/x/tools v0.1.2/go.mod
h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
@@ -1315,6 +1355,7 @@ rsc.io/sampler v1.3.0/go.mod
h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA=
sigs.k8s.io/yaml v1.1.0/go.mod h1:UJmg0vDUVViEyp3mgSv9WPwZCDxu4rQW1olrI1uml+o=
sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
sourcegraph.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0/go.mod
h1:hI742Nqp5OhwiqlzhgfbWU4mW4yO10fP+LoT9WOswdU=
+stathat.com/c/consistent v1.0.0/go.mod
h1:QkzMWzcbB+yQBL2AttO6sgsQS/JSTapcDISJalmCDS0=
vimagination.zapto.org/byteio v0.0.0-20200222190125-d27cba0f0b10
h1:pxt6fVJP67Hxo1qk8JalUghLlk3abYByl+3e0JYfUlE=
vimagination.zapto.org/byteio v0.0.0-20200222190125-d27cba0f0b10/go.mod
h1:fl9OF22g6MTKgvHA1hqMXe/L7+ULWofVTwbC9loGu7A=
vimagination.zapto.org/memio v0.0.0-20200222190306-588ebc67b97d
h1:Mp6WiHHuiwHaknxTdxJ8pvC9/B4pOgW1PamKGexG7Fs=
diff --git a/pkg/constant/context.go b/pkg/constant/context.go
index 5dd679c9..08db6633 100644
--- a/pkg/constant/context.go
+++ b/pkg/constant/context.go
@@ -39,4 +39,8 @@ const (
SeataVersion = "1.1.0"
TccBusinessActionContextParameter = "tccParam"
+
+ // RocketMQ message property keys for XID and BranchId propagation
+ PropertySeataXID = "SEATA_XID"
+ PropertySeataBranchId = "SEATA_BRANCH_ID"
)
diff --git a/pkg/integration/rocketmq/config.go
b/pkg/integration/rocketmq/config.go
new file mode 100644
index 00000000..e97e3b36
--- /dev/null
+++ b/pkg/integration/rocketmq/config.go
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rocketmq
+
+import (
+ "time"
+
+ "github.com/apache/rocketmq-client-go/v2/producer"
+)
+
+type SeataMQProducerConfig struct {
+ NameServerAddrs []string
+ GroupName string
+ Namespace string
+ InstanceName string
+
+ RetryTimesWhenSendFailed int
+ SendMsgTimeout time.Duration
+}
+
+func NewDefaultSeataMQProducerConfig() *SeataMQProducerConfig {
+ return &SeataMQProducerConfig{
+ RetryTimesWhenSendFailed: 3,
+ SendMsgTimeout: 3 * time.Second,
+ }
+}
+
+func (c *SeataMQProducerConfig) ToRocketMQProducerOptions() []producer.Option {
+ opts := []producer.Option{
+ producer.WithNameServer(c.NameServerAddrs),
+ producer.WithGroupName(c.GroupName),
+ producer.WithRetry(c.RetryTimesWhenSendFailed),
+ producer.WithSendMsgTimeout(c.SendMsgTimeout),
+ }
+ if c.Namespace != "" {
+ opts = append(opts, producer.WithNamespace(c.Namespace))
+ }
+ if c.InstanceName != "" {
+ opts = append(opts, producer.WithInstanceName(c.InstanceName))
+ }
+ return opts
+}
diff --git a/pkg/protocol/message/other_message_test.go
b/pkg/integration/rocketmq/constants.go
similarity index 60%
copy from pkg/protocol/message/other_message_test.go
copy to pkg/integration/rocketmq/constants.go
index d4487cc4..b0bbbe7f 100644
--- a/pkg/protocol/message/other_message_test.go
+++ b/pkg/integration/rocketmq/constants.go
@@ -15,24 +15,15 @@
* limitations under the License.
*/
-package message
+package rocketmq
-import (
- "testing"
+const (
+ ResourceIDTCCRocketMQ = "tccRocketMQ"
- "github.com/stretchr/testify/assert"
+ ActionContextKeyOffsetMsgId = "offsetMsgId"
+ ActionContextKeyMsgId = "msgId"
+ ActionContextKeyQueueId = "queueId"
+ ActionContextKeyQueueOffset = "queueOffset"
+ ActionContextKeyTransactionId = "transactionId"
+ ActionContextKeyBrokerName = "brokerName"
)
-
-func TestNewMessageFuture(t *testing.T) {
- rpcMessage := RpcMessage{ID: 0}
- assert.Equal(t, int32(0), NewMessageFuture(rpcMessage).ID)
-}
-
-func TestHeartBeatMessage_ToString(t *testing.T) {
- assert.Equal(t, "services ping", HeartBeatMessagePing.ToString())
- assert.Equal(t, "services pong", HeartBeatMessagePong.ToString())
-}
-
-func TestHeartBeatMessage_GetTypeCode(t *testing.T) {
- assert.Equal(t, MessageTypeHeartbeatMsg,
HeartBeatMessage{}.GetTypeCode())
-}
diff --git a/pkg/integration/rocketmq/producer_factory.go
b/pkg/integration/rocketmq/producer_factory.go
new file mode 100644
index 00000000..77a76283
--- /dev/null
+++ b/pkg/integration/rocketmq/producer_factory.go
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rocketmq
+
+import (
+ "fmt"
+ "sync"
+)
+
+var (
+ globalProducer *SeataMQProducer
+ producerMutex sync.RWMutex
+ producerInitialized bool
+ newSeataMQProducer = NewSeataMQProducer
+)
+
+func InitSeataMQProducer(cfg *SeataMQProducerConfig) error {
+ producerMutex.Lock()
+ defer producerMutex.Unlock()
+
+ // Allow re-initialization if previous attempt failed
+ if producerInitialized && globalProducer != nil {
+ return nil
+ }
+
+ producer, err := newSeataMQProducer(cfg)
+ if err != nil {
+ return fmt.Errorf("failed to create SeataMQProducer: %w", err)
+ }
+
+ if err := producer.Start(); err != nil {
+ return fmt.Errorf("failed to start SeataMQProducer: %w", err)
+ }
+
+ globalProducer = producer
+ producerInitialized = true
+
+ return nil
+}
+
+func GetSeataMQProducer() (*SeataMQProducer, error) {
+ producerMutex.RLock()
+ defer producerMutex.RUnlock()
+
+ if !producerInitialized || globalProducer == nil {
+ return nil, fmt.Errorf("SeataMQProducer not initialized, call
InitSeataMQProducer first")
+ }
+
+ return globalProducer, nil
+}
+
+func ShutdownSeataMQProducer() error {
+ producerMutex.Lock()
+ defer producerMutex.Unlock()
+
+ if !producerInitialized || globalProducer == nil {
+ return nil
+ }
+
+ err := globalProducer.Shutdown()
+ globalProducer = nil
+ producerInitialized = false
+
+ return err
+}
diff --git a/pkg/integration/rocketmq/producer_factory_test.go
b/pkg/integration/rocketmq/producer_factory_test.go
new file mode 100644
index 00000000..cffa8e8d
--- /dev/null
+++ b/pkg/integration/rocketmq/producer_factory_test.go
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rocketmq
+
+import (
+ "errors"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+func resetProducerFactoryForTest() {
+ globalProducer = nil
+ producerInitialized = false
+ newSeataMQProducer = NewSeataMQProducer
+}
+
+func newStubFactoryProducer() (*SeataMQProducer, *stubTransactionProducer,
*stubNormalProducer) {
+ transactionProducer := &stubTransactionProducer{}
+ normalProducer := &stubNormalProducer{}
+ return &SeataMQProducer{
+ transactionProducer: transactionProducer,
+ normalProducer: normalProducer,
+ }, transactionProducer, normalProducer
+}
+
+func TestGetSeataMQProducer_RequiresInitialization(t *testing.T) {
+ resetProducerFactoryForTest()
+ t.Cleanup(resetProducerFactoryForTest)
+
+ producer, err := GetSeataMQProducer()
+
+ require.Error(t, err)
+ assert.Nil(t, producer)
+ assert.Contains(t, err.Error(), "not initialized")
+}
+
+func TestInitSeataMQProducer_AllowsRetryAfterCreateFailure(t *testing.T) {
+ resetProducerFactoryForTest()
+ t.Cleanup(resetProducerFactoryForTest)
+
+ expectedErr := errors.New("create failed")
+ healthyProducer, transactionProducer, normalProducer :=
newStubFactoryProducer()
+ createCalls := 0
+ newSeataMQProducer = func(*SeataMQProducerConfig) (*SeataMQProducer,
error) {
+ createCalls++
+ if createCalls == 1 {
+ return nil, expectedErr
+ }
+ return healthyProducer, nil
+ }
+
+ err := InitSeataMQProducer(&SeataMQProducerConfig{})
+ require.ErrorIs(t, err, expectedErr)
+
+ got, getErr := GetSeataMQProducer()
+ require.Error(t, getErr)
+ assert.Nil(t, got)
+
+ err = InitSeataMQProducer(&SeataMQProducerConfig{})
+ require.NoError(t, err)
+
+ got, err = GetSeataMQProducer()
+ require.NoError(t, err)
+ assert.Same(t, healthyProducer, got)
+ assert.Equal(t, 2, createCalls)
+ assert.Equal(t, 1, transactionProducer.startCalls)
+ assert.Equal(t, 1, normalProducer.startCalls)
+}
+
+func TestInitSeataMQProducer_AllowsRetryAfterStartFailure(t *testing.T) {
+ resetProducerFactoryForTest()
+ t.Cleanup(resetProducerFactoryForTest)
+
+ expectedErr := errors.New("normal start failed")
+ failingProducer, failingTransactionProducer, failingNormalProducer :=
newStubFactoryProducer()
+ failingNormalProducer.startErr = expectedErr
+ healthyProducer, healthyTransactionProducer, healthyNormalProducer :=
newStubFactoryProducer()
+
+ createCalls := 0
+ newSeataMQProducer = func(*SeataMQProducerConfig) (*SeataMQProducer,
error) {
+ createCalls++
+ if createCalls == 1 {
+ return failingProducer, nil
+ }
+ return healthyProducer, nil
+ }
+
+ err := InitSeataMQProducer(&SeataMQProducerConfig{})
+ require.ErrorIs(t, err, expectedErr)
+ assert.Equal(t, 1, failingTransactionProducer.startCalls)
+ assert.Equal(t, 1, failingNormalProducer.startCalls)
+ assert.Equal(t, 1, failingTransactionProducer.shutdownCalls)
+
+ got, getErr := GetSeataMQProducer()
+ require.Error(t, getErr)
+ assert.Nil(t, got)
+
+ err = InitSeataMQProducer(&SeataMQProducerConfig{})
+ require.NoError(t, err)
+
+ got, err = GetSeataMQProducer()
+ require.NoError(t, err)
+ assert.Same(t, healthyProducer, got)
+ assert.Equal(t, 2, createCalls)
+ assert.Equal(t, 1, healthyTransactionProducer.startCalls)
+ assert.Equal(t, 1, healthyNormalProducer.startCalls)
+}
+
+func TestInitSeataMQProducer_DoesNotReinitializeAfterSuccess(t *testing.T) {
+ resetProducerFactoryForTest()
+ t.Cleanup(resetProducerFactoryForTest)
+
+ healthyProducer, transactionProducer, normalProducer :=
newStubFactoryProducer()
+ createCalls := 0
+ newSeataMQProducer = func(*SeataMQProducerConfig) (*SeataMQProducer,
error) {
+ createCalls++
+ return healthyProducer, nil
+ }
+
+ require.NoError(t, InitSeataMQProducer(&SeataMQProducerConfig{}))
+ require.NoError(t, InitSeataMQProducer(&SeataMQProducerConfig{}))
+
+ got, err := GetSeataMQProducer()
+ require.NoError(t, err)
+ assert.Same(t, healthyProducer, got)
+ assert.Equal(t, 1, createCalls)
+ assert.Equal(t, 1, transactionProducer.startCalls)
+ assert.Equal(t, 1, normalProducer.startCalls)
+}
+
+func TestShutdownSeataMQProducer_ResetsFactoryState(t *testing.T) {
+ resetProducerFactoryForTest()
+ t.Cleanup(resetProducerFactoryForTest)
+
+ healthyProducer, transactionProducer, normalProducer :=
newStubFactoryProducer()
+ newSeataMQProducer = func(*SeataMQProducerConfig) (*SeataMQProducer,
error) {
+ return healthyProducer, nil
+ }
+
+ require.NoError(t, InitSeataMQProducer(&SeataMQProducerConfig{}))
+ require.NoError(t, ShutdownSeataMQProducer())
+ require.NoError(t, ShutdownSeataMQProducer())
+
+ assert.Equal(t, 1, transactionProducer.shutdownCalls)
+ assert.Equal(t, 1, normalProducer.shutdownCalls)
+ assert.True(t, healthyProducer.closed)
+ assert.False(t, producerInitialized)
+ assert.Nil(t, globalProducer)
+
+ got, err := GetSeataMQProducer()
+ require.Error(t, err)
+ assert.Nil(t, got)
+}
diff --git a/pkg/integration/rocketmq/seata_producer.go
b/pkg/integration/rocketmq/seata_producer.go
new file mode 100644
index 00000000..3a22f8e6
--- /dev/null
+++ b/pkg/integration/rocketmq/seata_producer.go
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rocketmq
+
+import (
+ "context"
+ "fmt"
+ "sync"
+
+ "github.com/apache/rocketmq-client-go/v2/primitive"
+ "github.com/apache/rocketmq-client-go/v2/producer"
+
+ "seata.apache.org/seata-go/v2/pkg/rm/tcc"
+ "seata.apache.org/seata-go/v2/pkg/tm"
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+type transactionProducerInterface interface {
+ Start() error
+ Shutdown() error
+ SendMessageInTransaction(context.Context, *primitive.Message)
(*primitive.TransactionSendResult, error)
+}
+
+type normalProducerInterface interface {
+ Start() error
+ Shutdown() error
+ SendSync(ctx context.Context, msg ...*primitive.Message)
(*primitive.SendResult, error)
+}
+
+type tccPrepareProxy interface {
+ Prepare(context.Context, interface{}) (interface{}, error)
+}
+
+type SeataMQProducer struct {
+ config *SeataMQProducerConfig
+ transactionProducer transactionProducerInterface
+ normalProducer normalProducerInterface
+ tccAction *TCCRocketMQAction
+ tccProxy tccPrepareProxy
+
+ mu sync.RWMutex
+ closed bool
+}
+
+func NewSeataMQProducer(cfg *SeataMQProducerConfig) (*SeataMQProducer, error) {
+ if cfg == nil {
+ return nil, fmt.Errorf("config cannot be nil")
+ }
+
+ if cfg.NameServerAddrs == nil || len(cfg.NameServerAddrs) == 0 {
+ return nil, fmt.Errorf("NameServerAddrs cannot be empty")
+ }
+
+ if cfg.GroupName == "" {
+ return nil, fmt.Errorf("GroupName cannot be empty")
+ }
+
+ p := &SeataMQProducer{
+ config: cfg,
+ }
+
+ p.tccAction = NewTCCRocketMQAction(p)
+
+ // NewTCCServiceProxy internally calls ParseTCCResource, so we pass the
action directly
+ var err error
+ p.tccProxy, err = tcc.NewTCCServiceProxy(p.tccAction)
+ if err != nil {
+ return nil, fmt.Errorf("create TCC proxy failed: %w", err)
+ }
+
+ listener := NewSeataTransactionListener(p)
+ opts := cfg.ToRocketMQProducerOptions()
+
+ p.transactionProducer, err = producer.NewTransactionProducer(listener,
opts...)
+ if err != nil {
+ return nil, fmt.Errorf("create transaction producer failed:
%w", err)
+ }
+
+ normalCfg := *cfg
+ normalCfg.GroupName = cfg.GroupName + "-normal"
+ normalOpts := normalCfg.ToRocketMQProducerOptions()
+ p.normalProducer, err = producer.NewDefaultProducer(normalOpts...)
+ if err != nil {
+ return nil, fmt.Errorf("create normal producer failed: %w", err)
+ }
+
+ return p, nil
+}
+
+func (p *SeataMQProducer) Start() error {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ if p.closed {
+ return fmt.Errorf("producer already closed")
+ }
+
+ if err := p.transactionProducer.Start(); err != nil {
+ return err
+ }
+
+ if err := p.normalProducer.Start(); err != nil {
+ p.transactionProducer.Shutdown()
+ return err
+ }
+
+ return nil
+}
+
+func (p *SeataMQProducer) Shutdown() error {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+
+ if p.closed {
+ return nil
+ }
+
+ p.closed = true
+
+ var errs []error
+ if err := p.transactionProducer.Shutdown(); err != nil {
+ errs = append(errs, err)
+ }
+ if err := p.normalProducer.Shutdown(); err != nil {
+ errs = append(errs, err)
+ }
+
+ if len(errs) > 0 {
+ return fmt.Errorf("shutdown errors: %v", errs)
+ }
+ return nil
+}
+
+func (p *SeataMQProducer) Send(ctx context.Context, msg *primitive.Message)
(*primitive.SendResult, error) {
+ p.mu.RLock()
+ defer p.mu.RUnlock()
+
+ if p.closed {
+ return nil, fmt.Errorf("producer is closed")
+ }
+
+ if msg == nil {
+ return nil, fmt.Errorf("message cannot be nil")
+ }
+
+ if !tm.IsGlobalTx(ctx) {
+ return p.sendSync(ctx, msg)
+ }
+
+ _, err := p.tccProxy.Prepare(ctx, msg)
+ if err != nil {
+ log.Errorf("[SeataMQProducer] Send in global tx failed, xid=%s,
err=%v", tm.GetXID(ctx), err)
+ return nil, err
+ }
+
+ bac := tm.GetBusinessActionContext(ctx)
+ if bac == nil || bac.ActionContext == nil {
+ return nil, fmt.Errorf("BusinessActionContext action context
not found after Prepare")
+ }
+ return &primitive.SendResult{
+ Status: primitive.SendOK,
+ MsgID: getStringFromMap(bac.ActionContext,
ActionContextKeyMsgId),
+ OffsetMsgID: getStringFromMap(bac.ActionContext,
ActionContextKeyOffsetMsgId),
+ }, nil
+}
+
+func (p *SeataMQProducer) sendSync(ctx context.Context, msg
*primitive.Message) (*primitive.SendResult, error) {
+ return p.normalProducer.SendSync(ctx, msg)
+}
+
+func getStringFromMap(m map[string]interface{}, key string) string {
+ if v, ok := m[key]; ok {
+ if s, ok := v.(string); ok {
+ return s
+ }
+ }
+ return ""
+}
diff --git a/pkg/integration/rocketmq/seata_producer_test.go
b/pkg/integration/rocketmq/seata_producer_test.go
new file mode 100644
index 00000000..113e1cb6
--- /dev/null
+++ b/pkg/integration/rocketmq/seata_producer_test.go
@@ -0,0 +1,177 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rocketmq
+
+import (
+ "context"
+ "errors"
+ "testing"
+
+ "github.com/apache/rocketmq-client-go/v2/primitive"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "seata.apache.org/seata-go/v2/pkg/tm"
+)
+
+type stubTCCPrepareProxy struct {
+ prepareFn func(context.Context, interface{}) (interface{}, error)
+
+ prepareCalls int
+ lastCtx context.Context
+ lastParams interface{}
+}
+
+func (s *stubTCCPrepareProxy) Prepare(ctx context.Context, params interface{})
(interface{}, error) {
+ s.prepareCalls++
+ s.lastCtx = ctx
+ s.lastParams = params
+ if s.prepareFn != nil {
+ return s.prepareFn(ctx, params)
+ }
+ return nil, nil
+}
+
+func TestSeataMQProducerSend_NonGlobalTransactionUsesNormalProducer(t
*testing.T) {
+ normalProducer := &stubNormalProducer{
+ sendResult: &primitive.SendResult{
+ Status: primitive.SendOK,
+ MsgID: "msg-1",
+ },
+ }
+ prepareProxy := &stubTCCPrepareProxy{}
+ producer := &SeataMQProducer{
+ normalProducer: normalProducer,
+ tccProxy: prepareProxy,
+ }
+ msg := primitive.NewMessage("topic-test", []byte("hello"))
+
+ result, err := producer.Send(context.Background(), msg)
+
+ require.NoError(t, err)
+ require.NotNil(t, result)
+ assert.Equal(t, primitive.SendOK, result.Status)
+ assert.Equal(t, "msg-1", result.MsgID)
+ assert.Equal(t, 1, normalProducer.sendCalls)
+ assert.Same(t, msg, normalProducer.lastMsgs[0])
+ assert.Zero(t, prepareProxy.prepareCalls)
+}
+
+func TestSeataMQProducerSend_GlobalTransactionUsesTCCProxy(t *testing.T) {
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, "xid-123")
+ tm.SetBusinessActionContext(ctx, &tm.BusinessActionContext{
+ ActionContext: map[string]interface{}{},
+ })
+
+ prepareProxy := &stubTCCPrepareProxy{
+ prepareFn: func(ctx context.Context, params interface{})
(interface{}, error) {
+ bac := tm.GetBusinessActionContext(ctx)
+ bac.ActionContext[ActionContextKeyMsgId] = "msg-2"
+ bac.ActionContext[ActionContextKeyOffsetMsgId] =
"offset-2"
+ return true, nil
+ },
+ }
+ producer := &SeataMQProducer{
+ tccProxy: prepareProxy,
+ }
+ msg := primitive.NewMessage("topic-test", []byte("hello"))
+
+ result, err := producer.Send(ctx, msg)
+
+ require.NoError(t, err)
+ require.NotNil(t, result)
+ assert.Equal(t, primitive.SendOK, result.Status)
+ assert.Equal(t, "msg-2", result.MsgID)
+ assert.Equal(t, "offset-2", result.OffsetMsgID)
+ assert.Equal(t, 1, prepareProxy.prepareCalls)
+ assert.Same(t, msg, prepareProxy.lastParams)
+}
+
+func TestSeataMQProducerSend_GlobalTransactionPrepareError(t *testing.T) {
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, "xid-123")
+ tm.SetBusinessActionContext(ctx, &tm.BusinessActionContext{
+ ActionContext: map[string]interface{}{},
+ })
+
+ expectedErr := errors.New("prepare failed")
+ producer := &SeataMQProducer{
+ tccProxy: &stubTCCPrepareProxy{
+ prepareFn: func(context.Context, interface{})
(interface{}, error) {
+ return nil, expectedErr
+ },
+ },
+ }
+
+ result, err := producer.Send(ctx, primitive.NewMessage("topic-test",
[]byte("hello")))
+
+ require.ErrorIs(t, err, expectedErr)
+ assert.Nil(t, result)
+}
+
+func
TestSeataMQProducerSend_GlobalTransactionRequiresActionContextAfterPrepare(t
*testing.T) {
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, "xid-123")
+
+ producer := &SeataMQProducer{
+ tccProxy: &stubTCCPrepareProxy{
+ prepareFn: func(context.Context, interface{})
(interface{}, error) {
+ return true, nil
+ },
+ },
+ }
+
+ result, err := producer.Send(ctx, primitive.NewMessage("topic-test",
[]byte("hello")))
+
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "BusinessActionContext")
+ assert.Nil(t, result)
+}
+
+func TestSeataMQProducerSend_RejectsNilMessageAndClosedProducer(t *testing.T) {
+ producer := &SeataMQProducer{}
+
+ result, err := producer.Send(context.Background(), nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "message cannot be nil")
+ assert.Nil(t, result)
+
+ producer.closed = true
+ result, err = producer.Send(context.Background(),
primitive.NewMessage("topic-test", []byte("hello")))
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "producer is closed")
+ assert.Nil(t, result)
+}
+
+func
TestSeataMQProducerStart_ShutsDownTransactionProducerWhenNormalProducerStartFails(t
*testing.T) {
+ expectedErr := errors.New("normal start failed")
+ transactionProducer := &stubTransactionProducer{}
+ normalProducer := &stubNormalProducer{startErr: expectedErr}
+ producer := &SeataMQProducer{
+ transactionProducer: transactionProducer,
+ normalProducer: normalProducer,
+ }
+
+ err := producer.Start()
+
+ require.ErrorIs(t, err, expectedErr)
+ assert.Equal(t, 1, transactionProducer.startCalls)
+ assert.Equal(t, 1, normalProducer.startCalls)
+ assert.Equal(t, 1, transactionProducer.shutdownCalls)
+}
diff --git a/pkg/integration/rocketmq/tcc_rocketmq_action.go
b/pkg/integration/rocketmq/tcc_rocketmq_action.go
new file mode 100644
index 00000000..a769646d
--- /dev/null
+++ b/pkg/integration/rocketmq/tcc_rocketmq_action.go
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rocketmq
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/apache/rocketmq-client-go/v2/primitive"
+
+ "seata.apache.org/seata-go/v2/pkg/constant"
+ "seata.apache.org/seata-go/v2/pkg/tm"
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+type TCCRocketMQAction struct {
+ producer *SeataMQProducer
+}
+
+func NewTCCRocketMQAction(producer *SeataMQProducer) *TCCRocketMQAction {
+ return &TCCRocketMQAction{
+ producer: producer,
+ }
+}
+
+func (a *TCCRocketMQAction) GetActionName() string {
+ return ResourceIDTCCRocketMQ
+}
+
+func (a *TCCRocketMQAction) Prepare(ctx context.Context, params interface{})
(bool, error) {
+ msg, ok := params.(*primitive.Message)
+ if !ok {
+ return false, fmt.Errorf("params must be *primitive.Message,
got %T", params)
+ }
+
+ bac := tm.GetBusinessActionContext(ctx)
+ if bac == nil {
+ return false, fmt.Errorf("BusinessActionContext not found in
context")
+ }
+
+ xid := tm.GetXID(ctx)
+ if xid == "" {
+ return false, fmt.Errorf("XID not found in context")
+ }
+ if bac.ActionContext == nil {
+ bac.ActionContext = make(map[string]interface{}, 6)
+ }
+
+ msg.WithProperty(constant.PropertySeataXID, xid)
+ msg.WithProperty(constant.PropertySeataBranchId, fmt.Sprintf("%d",
bac.BranchId))
+
+ result, err :=
a.producer.transactionProducer.SendMessageInTransaction(ctx, msg)
+ if err != nil {
+ log.Errorf("[TCCRocketMQ] Prepare failed, xid=%s, err=%v", xid,
err)
+ return false, err
+ }
+
+ bac.ActionContext[ActionContextKeyMsgId] = result.MsgID
+ bac.ActionContext[ActionContextKeyOffsetMsgId] = result.OffsetMsgID
+ bac.ActionContext[ActionContextKeyQueueOffset] = result.QueueOffset
+ bac.ActionContext[ActionContextKeyTransactionId] = result.TransactionID
+ if result.MessageQueue != nil {
+ bac.ActionContext[ActionContextKeyQueueId] =
result.MessageQueue.QueueId
+ bac.ActionContext[ActionContextKeyBrokerName] =
result.MessageQueue.BrokerName
+ }
+
+ log.Infof("[TCCRocketMQ] Prepare success, xid=%s, branchId=%d,
msgId=%s", xid, bac.BranchId, result.MsgID)
+
+ return true, nil
+}
+
+func (a *TCCRocketMQAction) Commit(ctx context.Context, bac
*tm.BusinessActionContext) (bool, error) {
+ // Commit is a no-op because RocketMQ transactional messages use a
check-back mechanism.
+ // When the global transaction commits, RocketMQ will invoke
CheckLocalTransaction
+ // via SeataTransactionListener to determine the final message
disposition.
+ // The message has already been sent to the broker during Prepare phase
with an
+ // initial state of UnknowState, pending the check-back resolution.
+ log.Infof("[TCCRocketMQ] Commit (no-op, rely on check-back), xid=%s,
branchId=%d", bac.Xid, bac.BranchId)
+ return true, nil
+}
+
+func (a *TCCRocketMQAction) Rollback(ctx context.Context, bac
*tm.BusinessActionContext) (bool, error) {
+ // Rollback is a no-op because RocketMQ transactional messages use a
check-back mechanism.
+ // When the global transaction rolls back, RocketMQ will invoke
CheckLocalTransaction
+ // via SeataTransactionListener, which queries the TC for the global
status and returns
+ // RollbackMessageState, causing the broker to discard the message.
+ log.Infof("[TCCRocketMQ] Rollback (no-op, rely on check-back), xid=%s,
branchId=%d", bac.Xid, bac.BranchId)
+ return true, nil
+}
diff --git a/pkg/integration/rocketmq/tcc_rocketmq_action_test.go
b/pkg/integration/rocketmq/tcc_rocketmq_action_test.go
new file mode 100644
index 00000000..b3e24267
--- /dev/null
+++ b/pkg/integration/rocketmq/tcc_rocketmq_action_test.go
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rocketmq
+
+import (
+ "context"
+ "testing"
+
+ "github.com/apache/rocketmq-client-go/v2/primitive"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "seata.apache.org/seata-go/v2/pkg/constant"
+ "seata.apache.org/seata-go/v2/pkg/rm/tcc"
+ "seata.apache.org/seata-go/v2/pkg/tm"
+)
+
+func TestTCCRocketMQAction_ParseTCCResource(t *testing.T) {
+ action := &TCCRocketMQAction{}
+
+ resource, err := tcc.ParseTCCResource(action)
+ assert.NoError(t, err)
+ assert.NotNil(t, resource)
+
+ assert.Equal(t, ResourceIDTCCRocketMQ, resource.GetResourceId())
+ assert.Equal(t, ResourceIDTCCRocketMQ, action.GetActionName())
+}
+
+func TestTCCRocketMQAction_GetActionName(t *testing.T) {
+ action := &TCCRocketMQAction{}
+ assert.Equal(t, ResourceIDTCCRocketMQ, action.GetActionName())
+}
+
+func TestTCCRocketMQActionPrepare_SetsMessagePropertiesAndActionContext(t
*testing.T) {
+ transactionProducer := &stubTransactionProducer{
+ sendResult: &primitive.TransactionSendResult{
+ SendResult: &primitive.SendResult{
+ MsgID: "msg-1",
+ OffsetMsgID: "offset-1",
+ QueueOffset: 11,
+ TransactionID: "tx-1",
+ MessageQueue: &primitive.MessageQueue{
+ BrokerName: "broker-a",
+ QueueId: 3,
+ },
+ },
+ },
+ }
+ action := &TCCRocketMQAction{
+ producer: &SeataMQProducer{
+ transactionProducer: transactionProducer,
+ },
+ }
+ ctx := tm.InitSeataContext(context.Background())
+ tm.SetXID(ctx, "xid-123")
+ tm.SetBusinessActionContext(ctx, &tm.BusinessActionContext{
+ BranchId: 1001,
+ ActionContext: map[string]interface{}{},
+ })
+ msg := primitive.NewMessage("topic-test", []byte("hello"))
+
+ ok, err := action.Prepare(ctx, msg)
+
+ require.NoError(t, err)
+ assert.True(t, ok)
+ assert.Equal(t, "xid-123", msg.GetProperty(constant.PropertySeataXID))
+ assert.Equal(t, "1001", msg.GetProperty(constant.PropertySeataBranchId))
+ bac := tm.GetBusinessActionContext(ctx)
+ require.NotNil(t, bac)
+ assert.Equal(t, "msg-1", bac.ActionContext[ActionContextKeyMsgId])
+ assert.Equal(t, "offset-1",
bac.ActionContext[ActionContextKeyOffsetMsgId])
+ assert.EqualValues(t, int64(11),
bac.ActionContext[ActionContextKeyQueueOffset])
+ assert.Equal(t, "tx-1",
bac.ActionContext[ActionContextKeyTransactionId])
+ assert.Equal(t, 3, bac.ActionContext[ActionContextKeyQueueId])
+ assert.Equal(t, "broker-a",
bac.ActionContext[ActionContextKeyBrokerName])
+ assert.Equal(t, 1, transactionProducer.sendCalls)
+ assert.Same(t, msg, transactionProducer.lastMsg)
+}
diff --git a/pkg/integration/rocketmq/test_helpers_test.go
b/pkg/integration/rocketmq/test_helpers_test.go
new file mode 100644
index 00000000..9682007d
--- /dev/null
+++ b/pkg/integration/rocketmq/test_helpers_test.go
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rocketmq
+
+import (
+ "context"
+
+ "github.com/apache/rocketmq-client-go/v2/primitive"
+)
+
+type stubTransactionProducer struct {
+ startErr error
+ shutdownErr error
+ sendErr error
+ sendResult *primitive.TransactionSendResult
+
+ startCalls int
+ shutdownCalls int
+ sendCalls int
+ lastCtx context.Context
+ lastMsg *primitive.Message
+}
+
+func (s *stubTransactionProducer) Start() error {
+ s.startCalls++
+ return s.startErr
+}
+
+func (s *stubTransactionProducer) Shutdown() error {
+ s.shutdownCalls++
+ return s.shutdownErr
+}
+
+func (s *stubTransactionProducer) SendMessageInTransaction(ctx
context.Context, msg *primitive.Message) (*primitive.TransactionSendResult,
error) {
+ s.sendCalls++
+ s.lastCtx = ctx
+ s.lastMsg = msg
+ if s.sendErr != nil {
+ return nil, s.sendErr
+ }
+ if s.sendResult == nil {
+ s.sendResult = &primitive.TransactionSendResult{SendResult:
&primitive.SendResult{}}
+ }
+ return s.sendResult, nil
+}
+
+type stubNormalProducer struct {
+ startErr error
+ shutdownErr error
+ sendErr error
+ sendResult *primitive.SendResult
+
+ startCalls int
+ shutdownCalls int
+ sendCalls int
+ lastCtx context.Context
+ lastMsgs []*primitive.Message
+}
+
+func (s *stubNormalProducer) Start() error {
+ s.startCalls++
+ return s.startErr
+}
+
+func (s *stubNormalProducer) Shutdown() error {
+ s.shutdownCalls++
+ return s.shutdownErr
+}
+
+func (s *stubNormalProducer) SendSync(ctx context.Context, msg
...*primitive.Message) (*primitive.SendResult, error) {
+ s.sendCalls++
+ s.lastCtx = ctx
+ s.lastMsgs = msg
+ if s.sendErr != nil {
+ return nil, s.sendErr
+ }
+ if s.sendResult == nil {
+ s.sendResult = &primitive.SendResult{}
+ }
+ return s.sendResult, nil
+}
diff --git a/pkg/integration/rocketmq/transaction_listener.go
b/pkg/integration/rocketmq/transaction_listener.go
new file mode 100644
index 00000000..036b1482
--- /dev/null
+++ b/pkg/integration/rocketmq/transaction_listener.go
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rocketmq
+
+import (
+ "fmt"
+
+ "github.com/apache/rocketmq-client-go/v2/primitive"
+
+ "seata.apache.org/seata-go/v2/pkg/constant"
+ "seata.apache.org/seata-go/v2/pkg/protocol/message"
+ "seata.apache.org/seata-go/v2/pkg/remoting/getty"
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+type SeataTransactionListener struct {
+ remotingClient globalStatusRequestSender
+}
+
+type globalStatusRequestSender interface {
+ SendSyncRequest(msg interface{}) (interface{}, error)
+}
+
+func NewSeataTransactionListener(_ *SeataMQProducer) *SeataTransactionListener
{
+ return &SeataTransactionListener{
+ remotingClient: getty.GetGettyRemotingClient(),
+ }
+}
+
+func (l *SeataTransactionListener) ExecuteLocalTransaction(msg
*primitive.Message) primitive.LocalTransactionState {
+ xid := msg.GetProperty(constant.PropertySeataXID)
+ if xid == "" {
+ return primitive.CommitMessageState
+ }
+ log.Debugf("[SeataTransactionListener] ExecuteLocalTransaction, xid=%s,
returning UnknownState", xid)
+ return primitive.UnknowState
+}
+
+func (l *SeataTransactionListener) CheckLocalTransaction(msgExt
*primitive.MessageExt) primitive.LocalTransactionState {
+ xid := msgExt.GetProperty(constant.PropertySeataXID)
+ if xid == "" {
+ log.Warnf("[SeataTransactionListener] CheckLocalTransaction:
missing XID, rollback")
+ return primitive.RollbackMessageState
+ }
+
+ branchIdStr := msgExt.GetProperty(constant.PropertySeataBranchId)
+ log.Infof("[SeataTransactionListener] CheckLocalTransaction, xid=%s,
branchId=%s", xid, branchIdStr)
+
+ globalStatus, err := l.queryGlobalStatus(xid)
+ if err != nil {
+ log.Errorf("[SeataTransactionListener] Query global status
failed, xid=%s, err=%v", xid, err)
+ return primitive.UnknowState
+ }
+
+ localTransactionState :=
mapGlobalStatusToLocalTransactionState(globalStatus)
+ switch localTransactionState {
+ case primitive.CommitMessageState:
+ log.Infof("[SeataTransactionListener] Global tx committed,
xid=%s", xid)
+ case primitive.RollbackMessageState:
+ log.Infof("[SeataTransactionListener] Global tx rollbacked,
xid=%s, status=%v", xid, globalStatus)
+ default:
+ log.Infof("[SeataTransactionListener] Global tx waiting for
final state, xid=%s, status=%v", xid, globalStatus)
+ }
+ return localTransactionState
+}
+
+func (l *SeataTransactionListener) queryGlobalStatus(xid string)
(message.GlobalStatus, error) {
+ req := message.GlobalStatusRequest{
+ AbstractGlobalEndRequest: message.AbstractGlobalEndRequest{
+ Xid: xid,
+ },
+ }
+ res, err := l.remotingClient.SendSyncRequest(req)
+ if err != nil {
+ return message.GlobalStatusUnKnown, err
+ }
+ gsResp, ok := res.(message.GlobalStatusResponse)
+ if !ok {
+ log.Errorf("[SeataTransactionListener] Invalid response type
for GetGlobalStatus, xid=%s", xid)
+ return message.GlobalStatusUnKnown, fmt.Errorf("invalid
response type: %T", res)
+ }
+ return gsResp.GlobalStatus, nil
+}
+
+func mapGlobalStatusToLocalTransactionState(globalStatus message.GlobalStatus)
primitive.LocalTransactionState {
+ // Finished only means the TC no longer manages the session, so it
cannot safely
+ // distinguish a late check on a committed transaction from a rollback
outcome.
+ switch globalStatus {
+ case message.GlobalStatusCommitted, message.GlobalStatusAsyncCommitting:
+ return primitive.CommitMessageState
+ case message.GlobalStatusRollbacked,
message.GlobalStatusTimeoutRollbacked, message.GlobalStatusRollbackFailed,
+ message.GlobalStatusTimeoutRollbackFailed,
message.GlobalStatusCommitFailed:
+ return primitive.RollbackMessageState
+ case message.GlobalStatusBegin, message.GlobalStatusCommitting,
message.GlobalStatusCommitRetrying,
+ message.GlobalStatusRollbacking,
message.GlobalStatusRollbackRetrying, message.GlobalStatusTimeoutRollbacking,
+ message.GlobalStatusTimeoutRollbackRetrying,
message.GlobalStatusFinished:
+ return primitive.UnknowState
+ default:
+ return primitive.UnknowState
+ }
+}
diff --git a/pkg/integration/rocketmq/transaction_listener_test.go
b/pkg/integration/rocketmq/transaction_listener_test.go
new file mode 100644
index 00000000..9fe6248d
--- /dev/null
+++ b/pkg/integration/rocketmq/transaction_listener_test.go
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rocketmq
+
+import (
+ "errors"
+ "testing"
+
+ "github.com/apache/rocketmq-client-go/v2/primitive"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "seata.apache.org/seata-go/v2/pkg/constant"
+ "seata.apache.org/seata-go/v2/pkg/protocol/message"
+)
+
+type stubGlobalStatusRequestSender struct {
+ response interface{}
+ err error
+ lastReq interface{}
+}
+
+func (s *stubGlobalStatusRequestSender) SendSyncRequest(msg interface{})
(interface{}, error) {
+ s.lastReq = msg
+ if s.err != nil {
+ return nil, s.err
+ }
+ return s.response, nil
+}
+
+func TestSeataTransactionListenerExecuteLocalTransaction(t *testing.T) {
+ listener := NewSeataTransactionListener(nil)
+
+ msgWithoutXID := primitive.NewMessage("topic-test", []byte("hello"))
+ assert.Equal(t, primitive.CommitMessageState,
listener.ExecuteLocalTransaction(msgWithoutXID))
+
+ msgWithXID := primitive.NewMessage("topic-test", []byte("hello"))
+ msgWithXID.WithProperty(constant.PropertySeataXID, "xid-123")
+ assert.Equal(t, primitive.UnknowState,
listener.ExecuteLocalTransaction(msgWithXID))
+}
+
+func TestSeataTransactionListenerCheckLocalTransaction_MissingXIDRollsBack(t
*testing.T) {
+ listener := NewSeataTransactionListener(nil)
+
+ state := listener.CheckLocalTransaction(&primitive.MessageExt{})
+
+ assert.Equal(t, primitive.RollbackMessageState, state)
+}
+
+func TestSeataTransactionListenerCheckLocalTransaction_StatusMapping(t
*testing.T) {
+ tests := []struct {
+ name string
+ globalStatus message.GlobalStatus
+ expected primitive.LocalTransactionState
+ }{
+ {name: "committed", globalStatus:
message.GlobalStatusCommitted, expected: primitive.CommitMessageState},
+ {name: "async committing", globalStatus:
message.GlobalStatusAsyncCommitting, expected: primitive.CommitMessageState},
+ {name: "commit retrying", globalStatus:
message.GlobalStatusCommitRetrying, expected: primitive.UnknowState},
+ {name: "rollbacking", globalStatus:
message.GlobalStatusRollbacking, expected: primitive.UnknowState},
+ {name: "timeout rollback retrying", globalStatus:
message.GlobalStatusTimeoutRollbackRetrying, expected: primitive.UnknowState},
+ {name: "commit failed", globalStatus:
message.GlobalStatusCommitFailed, expected: primitive.RollbackMessageState},
+ {name: "timeout rollback failed", globalStatus:
message.GlobalStatusTimeoutRollbackFailed, expected:
primitive.RollbackMessageState},
+ {name: "finished", globalStatus: message.GlobalStatusFinished,
expected: primitive.UnknowState},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ client := &stubGlobalStatusRequestSender{
+ response: message.GlobalStatusResponse{
+ AbstractGlobalEndResponse:
message.AbstractGlobalEndResponse{
+ GlobalStatus: tt.globalStatus,
+ },
+ },
+ }
+ listener := &SeataTransactionListener{
+ remotingClient: client,
+ }
+
+ state :=
listener.CheckLocalTransaction(newTransactionMessageExt("xid-123", "1001"))
+
+ assert.Equal(t, tt.expected, state)
+ })
+ }
+}
+
+func
TestSeataTransactionListenerCheckLocalTransaction_QueryErrorReturnsUnknown(t
*testing.T) {
+ listener := &SeataTransactionListener{
+ remotingClient: &stubGlobalStatusRequestSender{
+ err: errors.New("network timeout"),
+ },
+ }
+
+ state :=
listener.CheckLocalTransaction(newTransactionMessageExt("xid-123", "1001"))
+
+ assert.Equal(t, primitive.UnknowState, state)
+}
+
+func TestSeataTransactionListenerQueryGlobalStatus(t *testing.T) {
+ client := &stubGlobalStatusRequestSender{
+ response: message.GlobalStatusResponse{
+ AbstractGlobalEndResponse:
message.AbstractGlobalEndResponse{
+ GlobalStatus: message.GlobalStatusRollbacked,
+ },
+ },
+ }
+ listener := &SeataTransactionListener{
+ remotingClient: client,
+ }
+
+ status, err := listener.queryGlobalStatus("xid-123")
+
+ require.NoError(t, err)
+ assert.Equal(t, message.GlobalStatusRollbacked, status)
+ req, ok := client.lastReq.(message.GlobalStatusRequest)
+ require.True(t, ok)
+ assert.Equal(t, "xid-123", req.Xid)
+}
+
+func TestSeataTransactionListenerQueryGlobalStatus_InvalidResponseType(t
*testing.T) {
+ listener := &SeataTransactionListener{
+ remotingClient: &stubGlobalStatusRequestSender{
+ response: "invalid-response",
+ },
+ }
+
+ status, err := listener.queryGlobalStatus("xid-123")
+
+ require.Error(t, err)
+ assert.Equal(t, message.GlobalStatusUnKnown, status)
+ assert.Contains(t, err.Error(), "invalid response type")
+}
+
+func newTransactionMessageExt(xid, branchID string) *primitive.MessageExt {
+ msg := primitive.NewMessage("topic-test", []byte("hello"))
+ if xid != "" {
+ msg.WithProperty(constant.PropertySeataXID, xid)
+ }
+ if branchID != "" {
+ msg.WithProperty(constant.PropertySeataBranchId, branchID)
+ }
+ return &primitive.MessageExt{Message: *msg}
+}
diff --git a/pkg/protocol/message/other_message.go
b/pkg/protocol/message/other_message.go
index dd2abf4b..9e4ddafa 100644
--- a/pkg/protocol/message/other_message.go
+++ b/pkg/protocol/message/other_message.go
@@ -36,7 +36,7 @@ type MessageFuture struct {
func NewMessageFuture(message RpcMessage) *MessageFuture {
return &MessageFuture{
ID: message.ID,
- Done: make(chan struct{}),
+ Done: make(chan struct{}, 1), // Buffered channel prevents lost
signals in timeout race
}
}
diff --git a/pkg/protocol/message/other_message_test.go
b/pkg/protocol/message/other_message_test.go
index d4487cc4..5625ad4c 100644
--- a/pkg/protocol/message/other_message_test.go
+++ b/pkg/protocol/message/other_message_test.go
@@ -25,7 +25,9 @@ import (
func TestNewMessageFuture(t *testing.T) {
rpcMessage := RpcMessage{ID: 0}
- assert.Equal(t, int32(0), NewMessageFuture(rpcMessage).ID)
+ messageFuture := NewMessageFuture(rpcMessage)
+ assert.Equal(t, int32(0), messageFuture.ID)
+ assert.Equal(t, 1, cap(messageFuture.Done))
}
func TestHeartBeatMessage_ToString(t *testing.T) {
diff --git a/pkg/remoting/getty/getty_remoting.go
b/pkg/remoting/getty/getty_remoting.go
index ef0b419e..7b4e1e53 100644
--- a/pkg/remoting/getty/getty_remoting.go
+++ b/pkg/remoting/getty/getty_remoting.go
@@ -128,7 +128,11 @@ func (g *GettyRemoting)
NotifyRpcMessageResponse(rpcMessage message.RpcMessage)
messageFuture.Response = rpcMessage.Body
// todo add messageFuture.Err
// messageFuture.Err = rpcMessage.Err
- messageFuture.Done <- struct{}{}
+ select {
+ case messageFuture.Done <- struct{}{}:
+ default:
+ log.Warnf("response notification dropped for msg ID: %d
because the future was already signaled", rpcMessage.ID)
+ }
// client.msgFutures.Delete(rpcMessage.RequestID)
} else {
log.Infof("msg: {} is not found in msgFutures.", rpcMessage.ID)
diff --git a/pkg/remoting/getty/getty_remoting_test.go
b/pkg/remoting/getty/getty_remoting_test.go
index 1eadc4f2..303633bb 100644
--- a/pkg/remoting/getty/getty_remoting_test.go
+++ b/pkg/remoting/getty/getty_remoting_test.go
@@ -19,6 +19,7 @@ package getty
import (
"testing"
+ "time"
"github.com/stretchr/testify/assert"
@@ -165,3 +166,48 @@ func TestGettyRemoting_RemoveMergedMessageFuture(t
*testing.T) {
})
}
}
+
+func TestGettyRemoting_NotifyRpcMessageResponseSignalsWaitingFuture(t
*testing.T) {
+ gettyRemoting := newGettyRemoting()
+ request := message.RpcMessage{ID: 1}
+ messageFuture := message.NewMessageFuture(request)
+ gettyRemoting.futures.Store(request.ID, messageFuture)
+
+ gettyRemoting.NotifyRpcMessageResponse(message.RpcMessage{
+ ID: request.ID,
+ Body: "ok",
+ })
+
+ assert.Equal(t, "ok", messageFuture.Response)
+ select {
+ case <-messageFuture.Done:
+ default:
+ t.Fatal("expected NotifyRpcMessageResponse to signal the
waiting future")
+ }
+}
+
+func
TestGettyRemoting_NotifyRpcMessageResponseDoesNotBlockWhenAlreadySignaled(t
*testing.T) {
+ gettyRemoting := newGettyRemoting()
+ request := message.RpcMessage{ID: 1}
+ messageFuture := message.NewMessageFuture(request)
+ messageFuture.Done <- struct{}{}
+ gettyRemoting.futures.Store(request.ID, messageFuture)
+
+ done := make(chan struct{})
+ go func() {
+ gettyRemoting.NotifyRpcMessageResponse(message.RpcMessage{
+ ID: request.ID,
+ Body: "late-response",
+ })
+ close(done)
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(time.Second):
+ t.Fatal("NotifyRpcMessageResponse blocked when the future was
already signaled")
+ }
+
+ assert.Equal(t, "late-response", messageFuture.Response)
+ assert.Len(t, messageFuture.Done, 1)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
