opensource/dtm
3
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

change tcc and xa func to return resty.Response

Browse Source
This commit is contained in:
yedf2 2021-08-03 09:48:24 +08:00
parent 84860ebda3
commit eb4a4cde60
8 changed files with 20 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dtmcli/tcc.go
View File

@ -16,7 +16,7 @@ type Tcc struct {
} }
// TccGlobalFunc type of global tcc call // TccGlobalFunc type of global tcc call
type TccGlobalFunc func(tcc *Tcc) (interface{}, error) type TccGlobalFunc func(tcc *Tcc) (*resty.Response, error)
// TccGlobalTransaction begin a tcc global transaction // TccGlobalTransaction begin a tcc global transaction
// dtm dtm服务器地址 // dtm dtm服务器地址

2
dtmcli/xa.go
View File

@ -16,7 +16,7 @@ type M = map[string]interface{}
var e2p = common.E2P var e2p = common.E2P
// XaGlobalFunc type of xa global function // XaGlobalFunc type of xa global function
type XaGlobalFunc func(xa *Xa) (interface{}, error) type XaGlobalFunc func(xa *Xa) (*resty.Response, error)
// XaLocalFunc type of xa local function // XaLocalFunc type of xa local function
type XaLocalFunc func(db *sql.DB, xa *Xa) (interface{}, error) type XaLocalFunc func(db *sql.DB, xa *Xa) (interface{}, error)

7
dtmsvr/trans_tcc_barrier_test.go
View File

@ -7,6 +7,7 @@ import (
"time" "time"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"github.com/go-resty/resty/v2"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/yedf/dtm/common" "github.com/yedf/dtm/common"
@ -23,7 +24,7 @@ func TestTccBarrier(t *testing.T) {
func tccBarrierRollback(t *testing.T) { func tccBarrierRollback(t *testing.T) {
gid := "tccBarrierRollback" gid := "tccBarrierRollback"
resp, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { resp, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) {
resp, err := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel") resp, err := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel")
assert.True(t, !dtmcli.IsFailure(resp, err)) assert.True(t, !dtmcli.IsFailure(resp, err))
return tcc.CallBranch(&examples.TransReq{Amount: 30, TransInResult: "FAILURE"}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInCancel") return tcc.CallBranch(&examples.TransReq{Amount: 30, TransInResult: "FAILURE"}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInCancel")
@ -35,7 +36,7 @@ func tccBarrierRollback(t *testing.T) {
func tccBarrierNormal(t *testing.T) { func tccBarrierNormal(t *testing.T) {
gid := "tccBarrierNormal" gid := "tccBarrierNormal"
resp, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { resp, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) {
resp, err := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel") resp, err := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel")
assert.True(t, !dtmcli.IsFailure(resp, err)) assert.True(t, !dtmcli.IsFailure(resp, err))
return tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInCancel") return tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInCancel")
@ -49,7 +50,7 @@ func tccBarrierDisorder(t *testing.T) {
timeoutChan := make(chan string, 2) timeoutChan := make(chan string, 2)
finishedChan := make(chan string, 2) finishedChan := make(chan string, 2)
gid := "tccBarrierDisorder" gid := "tccBarrierDisorder"
_, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { _, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) {
body := &examples.TransReq{Amount: 30} body := &examples.TransReq{Amount: 30}
tryURL := Busi + "/TccBTransOutTry" tryURL := Busi + "/TccBTransOutTry"
confirmURL := Busi + "/TccBTransOutConfirm" confirmURL := Busi + "/TccBTransOutConfirm"

5
dtmsvr/trans_tcc_test.go
View File

@ -4,6 +4,7 @@ import (
"testing" "testing"
"time" "time"
"github.com/go-resty/resty/v2"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/yedf/dtm/dtmcli" "github.com/yedf/dtm/dtmcli"
"github.com/yedf/dtm/examples" "github.com/yedf/dtm/examples"
@ -18,7 +19,7 @@ func TestTcc(t *testing.T) {
func tccNormal(t *testing.T) { func tccNormal(t *testing.T) {
data := &examples.TransReq{Amount: 30} data := &examples.TransReq{Amount: 30}
gid := "tccNormal" gid := "tccNormal"
ret, err := dtmcli.TccGlobalTransaction(examples.DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { ret, err := dtmcli.TccGlobalTransaction(examples.DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) {
resp, err := tcc.CallBranch(data, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert") resp, err := tcc.CallBranch(data, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert")
if dtmcli.IsFailure(resp, err) { if dtmcli.IsFailure(resp, err) {
return resp, err return resp, err
@ -31,7 +32,7 @@ func tccNormal(t *testing.T) {
func tccRollback(t *testing.T) { func tccRollback(t *testing.T) {
gid := "tccRollback" gid := "tccRollback"
data := &examples.TransReq{Amount: 30, TransInResult: "FAILURE"} data := &examples.TransReq{Amount: 30, TransInResult: "FAILURE"}
resp, err := dtmcli.TccGlobalTransaction(examples.DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { resp, err := dtmcli.TccGlobalTransaction(examples.DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) {
resp, rerr := tcc.CallBranch(data, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert") resp, rerr := tcc.CallBranch(data, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert")
assert.True(t, !dtmcli.IsFailure(resp, rerr)) assert.True(t, !dtmcli.IsFailure(resp, rerr))
examples.MainSwitch.TransOutRevertResult.SetOnce("PENDING") examples.MainSwitch.TransOutRevertResult.SetOnce("PENDING")

7
dtmsvr/trans_xa_test.go
View File

@ -4,6 +4,7 @@ import (
"fmt" "fmt"
"testing" "testing"
"github.com/go-resty/resty/v2"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/yedf/dtm/dtmcli" "github.com/yedf/dtm/dtmcli"
"github.com/yedf/dtm/examples" "github.com/yedf/dtm/examples"
@ -19,7 +20,7 @@ func TestXa(t *testing.T) {
} }
func xaLocalError(t *testing.T) { func xaLocalError(t *testing.T) {
_, err := examples.XaClient.XaGlobalTransaction("xaLocalError", func(xa *dtmcli.Xa) (interface{}, error) { _, err := examples.XaClient.XaGlobalTransaction("xaLocalError", func(xa *dtmcli.Xa) (*resty.Response, error) {
return nil, fmt.Errorf("an error") return nil, fmt.Errorf("an error")
}) })
assert.Error(t, err, fmt.Errorf("an error")) assert.Error(t, err, fmt.Errorf("an error"))
@ -28,7 +29,7 @@ func xaLocalError(t *testing.T) {
func xaNormal(t *testing.T) { func xaNormal(t *testing.T) {
xc := examples.XaClient xc := examples.XaClient
gid := "xaNormal" gid := "xaNormal"
res, err := xc.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (interface{}, error) { res, err := xc.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (*resty.Response, error) {
req := examples.GenTransReq(30, false, false) req := examples.GenTransReq(30, false, false)
resp, err := xa.CallBranch(req, examples.Busi+"/TransOutXa") resp, err := xa.CallBranch(req, examples.Busi+"/TransOutXa")
if dtmcli.IsFailure(resp, err) { if dtmcli.IsFailure(resp, err) {
@ -44,7 +45,7 @@ func xaNormal(t *testing.T) {
func xaRollback(t *testing.T) { func xaRollback(t *testing.T) {
xc := examples.XaClient xc := examples.XaClient
gid := "xaRollback" gid := "xaRollback"
res, err := xc.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (interface{}, error) { res, err := xc.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (*resty.Response, error) {
req := &examples.TransReq{Amount: 30, TransInResult: "FAILURE"} req := &examples.TransReq{Amount: 30, TransInResult: "FAILURE"}
resp, err := xa.CallBranch(req, examples.Busi+"/TransOutXa") resp, err := xa.CallBranch(req, examples.Busi+"/TransOutXa")
if dtmcli.IsFailure(resp, err) { if dtmcli.IsFailure(resp, err) {

5
examples/main_tcc.go
View File

@ -2,6 +2,7 @@ package examples
import ( import (
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"github.com/go-resty/resty/v2"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
"github.com/yedf/dtm/common" "github.com/yedf/dtm/common"
"github.com/yedf/dtm/dtmcli" "github.com/yedf/dtm/dtmcli"
@ -20,7 +21,7 @@ func TccSetup(app *gin.Engine) {
// TccFireRequestNested 1 // TccFireRequestNested 1
func TccFireRequestNested() string { func TccFireRequestNested() string {
gid := dtmcli.MustGenGid(DtmServer) gid := dtmcli.MustGenGid(DtmServer)
ret, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { ret, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) {
resp, err := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert") resp, err := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert")
if dtmcli.IsFailure(resp, err) { if dtmcli.IsFailure(resp, err) {
return resp, err return resp, err
@ -35,7 +36,7 @@ func TccFireRequestNested() string {
func TccFireRequest() string { func TccFireRequest() string {
logrus.Printf("tcc simple transaction begin") logrus.Printf("tcc simple transaction begin")
gid := dtmcli.MustGenGid(DtmServer) gid := dtmcli.MustGenGid(DtmServer)
ret, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { ret, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) {
resp, err := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert") resp, err := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert")
if dtmcli.IsFailure(resp, err) { if dtmcli.IsFailure(resp, err) {
return resp, err return resp, err

3
examples/main_tcc_barrier.go
View File

@ -5,6 +5,7 @@ import (
"fmt" "fmt"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"github.com/go-resty/resty/v2"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
"github.com/yedf/dtm/common" "github.com/yedf/dtm/common"
"github.com/yedf/dtm/dtmcli" "github.com/yedf/dtm/dtmcli"
@ -14,7 +15,7 @@ import (
func TccBarrierFireRequest() string { func TccBarrierFireRequest() string {
logrus.Printf("tcc transaction begin") logrus.Printf("tcc transaction begin")
gid := dtmcli.MustGenGid(DtmServer) gid := dtmcli.MustGenGid(DtmServer)
ret, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (interface{}, error) { ret, err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) {
resp, err := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel") resp, err := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel")
if dtmcli.IsFailure(resp, err) { if dtmcli.IsFailure(resp, err) {
return resp, err return resp, err

3
examples/main_xa.go
View File

@ -4,6 +4,7 @@ import (
"database/sql" "database/sql"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"github.com/go-resty/resty/v2"
"github.com/yedf/dtm/common" "github.com/yedf/dtm/common"
"github.com/yedf/dtm/dtmcli" "github.com/yedf/dtm/dtmcli"
) )
@ -27,7 +28,7 @@ func XaSetup(app *gin.Engine) {
// XaFireRequest 注册全局XA事务，调用XA的分支 // XaFireRequest 注册全局XA事务，调用XA的分支
func XaFireRequest() string { func XaFireRequest() string {
gid := dtmcli.MustGenGid(DtmServer) gid := dtmcli.MustGenGid(DtmServer)
res, err := XaClient.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (interface{}, error) { res, err := XaClient.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (*resty.Response, error) {
resp, err := xa.CallBranch(&TransReq{Amount: 30}, Busi+"/TransOutXa") resp, err := xa.CallBranch(&TransReq{Amount: 30}, Busi+"/TransOutXa")
if dtmcli.IsFailure(resp, err) { if dtmcli.IsFailure(resp, err) {
return resp, err return resp, err