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

test cover more

Browse Source
This commit is contained in:
yedongfu 2021-07-16 18:17:45 +08:00
parent 28b8f91969
commit a920fbadc3
4 changed files with 62 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dtmcli/tcc.go
View File

@ -26,7 +26,7 @@ func TccGlobalTransaction(dtm string, tccFunc TccGlobalFunc) (gid string, rerr e
"trans_type": "tcc",
}
defer func() {
if x := recover(); x != nil {
if x := recover(); x != nil || rerr != nil {
_, rerr = common.RestyClient.R().SetBody(data).Post(dtm + "/abort")
} else {
_, rerr = common.RestyClient.R().SetBody(data).Post(dtm + "/submit")

41
dtmsvr/dtmsvr_test.go
View File

@ -3,6 +3,7 @@ package dtmsvr
import (
"database/sql"
"fmt"
"strings"
"testing"
"time"
@ -42,7 +43,9 @@ func TestDtmSvr(t *testing.T) {
examples.ResetXaData()
tccBarrierNormal(t)
tccBarrierRollback(t)
sagaBarrierNormal(t)
sagaBarrierRollback(t)
msgNormal(t)
msgPending(t)
tccNormal(t)
@ -139,12 +142,12 @@ func tccNormal(t *testing.T) {
}
func tccBarrierNormal(t *testing.T) {
_, err := dtmcli.TccGlobalTransaction(DtmServer, func(tcc *dtmcli.Tcc) (rerr error) {
res1, rerr := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutRevert")
res1, rerr := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel")
e2p(rerr)
if res1.StatusCode() != 200 {
return fmt.Errorf("bad status code: %d", res1.StatusCode())
}
res2, rerr := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInRevert")
res2, rerr := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInCancel")
e2p(rerr)
if res2.StatusCode() != 200 {
return fmt.Errorf("bad status code: %d", res2.StatusCode())
@ -155,6 +158,29 @@ func tccBarrierNormal(t *testing.T) {
e2p(err)
}
func tccBarrierRollback(t *testing.T) {
gid, err := dtmcli.TccGlobalTransaction(DtmServer, func(tcc *dtmcli.Tcc) (rerr error) {
res1, rerr := tcc.CallBranch(&examples.TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel")
e2p(rerr)
if res1.StatusCode() != 200 {
return fmt.Errorf("bad status code: %d", res1.StatusCode())
}
res2, rerr := tcc.CallBranch(&examples.TransReq{Amount: 30, TransInResult: "FAIL"}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInCancel")
e2p(rerr)
if res2.StatusCode() != 200 {
return fmt.Errorf("bad status code: %d", res2.StatusCode())
}
if strings.Contains(res2.String(), "FAIL") {
return fmt.Errorf("branch trans in fail")
}
logrus.Printf("tcc returns: %s, %s", res1.String(), res2.String())
return
})
e2p(err)
WaitTransProcessed(gid)
assert.Equal(t, "failed", getTransStatus(gid))
}
func tccRollback(t *testing.T) {
data := &examples.TransReq{Amount: 30, TransInResult: "FAIL"}
_, err := dtmcli.TccGlobalTransaction(examples.DtmServer, func(tcc *dtmcli.Tcc) (rerr error) {
@ -219,6 +245,17 @@ func sagaRollback(t *testing.T) {
assert.Equal(t, []string{"succeed", "succeed", "succeed", "failed"}, getBranchesStatus(saga.Gid))
}
func sagaBarrierRollback(t *testing.T) {
saga := dtmcli.NewSaga(DtmServer).
Add(Busi+"/SagaBTransOut", Busi+"/SagaBTransOutCompensate", &examples.TransReq{Amount: 30}).
Add(Busi+"/SagaBTransIn", Busi+"/SagaBTransInCompensate", &examples.TransReq{Amount: 30, TransInResult: "FAIL"})
logrus.Printf("busi trans submit")
err := saga.Submit()
e2p(err)
WaitTransProcessed(saga.Gid)
assert.Equal(t, "failed", getTransStatus(saga.Gid))
}
func sagaCommittedPending(t *testing.T) {
saga := genSaga("gid-committedPending", false, false)
examples.MainSwitch.TransInResult.SetOnce("PENDING")

12
examples/main_saga_barrier.go
View File

@ -40,8 +40,12 @@ func sagaBarrierAdjustBalance(sdb *sql.DB, uid int, amount int) (interface{}, er
}
func sagaBarrierTransIn(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
if req.TransInResult != "" {
return req.TransInResult, nil
}
return dtmcli.ThroughBarrierCall(dbGet().ToSQLDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return sagaBarrierAdjustBalance(sdb, 1, reqFrom(c).Amount)
return sagaBarrierAdjustBalance(sdb, 1, req.Amount)
})
}
@ -52,8 +56,12 @@ func sagaBarrierTransInCompensate(c *gin.Context) (interface{}, error) {
}
func sagaBarrierTransOut(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
if req.TransInResult != "" {
return req.TransInResult, nil
}
return dtmcli.ThroughBarrierCall(dbGet().ToSQLDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return sagaBarrierAdjustBalance(sdb, 2, -reqFrom(c).Amount)
return sagaBarrierAdjustBalance(sdb, 2, -req.Amount)
})
}

16
examples/main_tcc_barrier.go
View File

@ -14,14 +14,14 @@ import (
func TccBarrierFireRequest() string {
logrus.Printf("tcc transaction begin")
gid, err := dtmcli.TccGlobalTransaction(DtmServer, func(tcc *dtmcli.Tcc) (rerr error) {
res1, rerr := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutRevert")
res1, rerr := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutCancel")
if rerr != nil {
return
}
if res1.StatusCode() != 200 {
return fmt.Errorf("bad status code: %d", res1.StatusCode())
}
res2, rerr := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInRevert")
res2, rerr := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInCancel")
if rerr != nil {
return
}
@ -75,8 +75,12 @@ func adjustBalance(sdb *sql.DB, uid int, amount int) (interface{}, error) {
// TCC下，转入
func tccBarrierTransInTry(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
if req.TransInResult != "" {
return req.TransInResult, nil
}
return dtmcli.ThroughBarrierCall(dbGet().ToSQLDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustTrading(sdb, transInUID, reqFrom(c).Amount)
return adjustTrading(sdb, transInUID, req.Amount)
})
}
@ -93,8 +97,12 @@ func tccBarrierTransInCancel(c *gin.Context) (interface{}, error) {
}
func tccBarrierTransOutTry(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
if req.TransInResult != "" {
return req.TransInResult, nil
}
return dtmcli.ThroughBarrierCall(dbGet().ToSQLDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustTrading(sdb, transOutUID, -reqFrom(c).Amount)
return adjustTrading(sdb, transOutUID, -req.Amount)
})
}