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

tcc pass compile

Browse Source
This commit is contained in:
yedongfu 2021-07-03 22:07:45 +08:00
parent 2a3297a3c5
commit ed4db3b854
7 changed files with 130 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
common/types.go
View File

@ -11,6 +11,9 @@ import (
"gorm.io/gorm"
)
type M = map[string]interface{}
type MS = map[string]string
type ModelBase struct {
ID uint
CreateTime *time.Time `gorm:"autoCreateTime"`

10
common/utils.go
View File

@ -23,8 +23,6 @@ import (
"github.com/spf13/viper"
)
type M = map[string]interface{}
func P2E(perr *error) {
if x := recover(); x != nil {
if e, ok := x.(error); ok {
@ -86,10 +84,6 @@ func MustAtoi(s string) int {
return r
}
func GenGid() string {
return getOneHexIp() + "-" + gNode.Generate().Base58()
}
var gNode *snowflake.Node = nil
func init() {
@ -98,6 +92,10 @@ func init() {
gNode = node
}
func GenGid() string {
return getOneHexIp() + "-" + gNode.Generate().Base58()
}
func OrString(ss ...string) string {
for _, s := range ss {
if s != "" {

87
dtmcli/tcc.go
View File

@ -3,57 +3,62 @@ package dtmcli
import (
"fmt"
jsonitor "github.com/json-iterator/go"
"github.com/sirupsen/logrus"
"github.com/gin-gonic/gin"
"github.com/go-resty/resty/v2"
"github.com/yedf/dtm/common"
)
type Tcc struct {
TccData
Server string
Dtm string
Gid string
}
type TccData struct {
Gid string `json:"gid"`
TransType string `json:"trans_type"`
Steps []TccStep `json:"steps"`
}
type TccStep struct {
Try string `json:"try"`
Confirm string `json:"confirm"`
Cancel string `json:"cancel"`
Data string `json:"data"`
type TccGlobalFunc func(tcc *Tcc) error
func TccGlobalTransaction(dtm string, tccFunc TccGlobalFunc) (gid string, rerr error) {
gid = common.GenGid()
data := &M{
"gid": gid,
"trans_type": "tcc",
}
defer func() {
if x := recover(); x != nil {
_, rerr = common.RestyClient.R().SetBody(data).Post(dtm + "/abort")
} else {
_, rerr = common.RestyClient.R().SetBody(data).Post(dtm + "/submit")
}
}()
tcc := &Tcc{Dtm: dtm, Gid: gid}
_, rerr = common.RestyClient.R().SetBody(data).Post(tcc.Dtm + "/prepare")
if rerr != nil {
return
}
rerr = tccFunc(tcc)
return
}
func NewTcc(server string) *Tcc {
return &Tcc{
TccData: TccData{
TransType: "tcc",
},
Server: server,
func TccFromReq(c *gin.Context) (*Tcc, error) {
tcc := &Tcc{
Dtm: c.Query("dtm"),
Gid: c.Query("gid"),
}
}
func (s *Tcc) Add(try string, confirm string, cancel string, data interface{}) *Tcc {
logrus.Printf("tcc %s Add %s %s %s %v", s.Gid, try, confirm, cancel, data)
step := TccStep{
Try: try,
Confirm: confirm,
Cancel: cancel,
Data: common.MustMarshalString(data),
if tcc.Dtm == "" || tcc.Gid == "" {
return nil, fmt.Errorf("bad tcc info. dtm: %s, gid: %s", tcc.Dtm, tcc.Gid)
}
s.Steps = append(s.Steps, step)
return s
return tcc, nil
}
func (s *Tcc) Submit() error {
logrus.Printf("committing %s body: %v", s.Gid, &s.TccData)
resp, err := common.RestyClient.R().SetBody(&s.TccData).Post(fmt.Sprintf("%s/submit", s.Server))
if err != nil {
return err
}
if resp.StatusCode() != 200 {
return fmt.Errorf("submit failed: %v", resp.Body())
}
s.Gid = jsonitor.Get(resp.Body(), "gid").ToString()
return nil
func (t *Tcc) CallBranch(body interface{}, tryUrl string, confirmUrl string, cancelUrl string) (*resty.Response, error) {
return common.RestyClient.R().
SetBody(&M{
"gid": t.Gid,
"branch": common.GenGid(),
"trans_type": "tcc",
"status": "prepared",
"data": string(common.MustMarshal(body)),
"try": tryUrl,
"confirm": confirmUrl,
"cancel": cancelUrl,
}).
Post(t.Dtm + "/registerXaBranch")
}

2
dtmcli/xa.go
View File

@ -65,7 +65,7 @@ func (xa *Xa) XaLocalTransaction(gid string, transFunc XaLocalFunc) (rerr error)
e2p(err)
resp, err := common.RestyClient.R().
SetBody(&M{"gid": gid, "branch": branch, "trans_type": "xa", "status": "prepared", "url": xa.CallbackUrl}).
Post(xa.Server + "/branch")
Post(xa.Server + "/registerXaBranch")
e2p(err)
if !strings.Contains(resp.String(), "SUCCESS") {
e2p(fmt.Errorf("unknown server response: %s", resp.String()))

29
dtmsvr/api.go
View File

@ -13,7 +13,8 @@ import (
func AddRoute(engine *gin.Engine) {
engine.POST("/api/dtmsvr/prepare", common.WrapHandler(Prepare))
engine.POST("/api/dtmsvr/submit", common.WrapHandler(Submit))
engine.POST("/api/dtmsvr/branch", common.WrapHandler(Branch))
engine.POST("/api/dtmsvr/registerXaBranch", common.WrapHandler(RegisterXaBranch))
engine.POST("/api/dtmsvr/registerTccBranch", common.WrapHandler(RegisterTccBranch))
engine.POST("/api/dtmsvr/abort", common.WrapHandler(Abort))
engine.GET("/api/dtmsvr/query", common.WrapHandler(Query))
}
@ -46,7 +47,7 @@ func Abort(c *gin.Context) (interface{}, error) {
return M{"message": "SUCCESS"}, nil
}
func Branch(c *gin.Context) (interface{}, error) {
func RegisterXaBranch(c *gin.Context) (interface{}, error) {
branch := TransBranch{}
err := c.BindJSON(&branch)
e2p(err)
@ -61,6 +62,30 @@ func Branch(c *gin.Context) (interface{}, error) {
return M{"message": "SUCCESS"}, nil
}
func RegisterTccBranch(c *gin.Context) (interface{}, error) {
data := common.MS{}
err := c.BindJSON(&data)
e2p(err)
branch := TransBranch{
Gid: data["gid"],
Branch: data["branch_id"],
Status: data["status"],
Data: data["data"],
}
branches := []*TransBranch{&branch, &branch, &branch}
for i, b := range []string{"cancel", "confirm", "try"} {
branches[i].BranchType = b
branches[i].Url = data[b]
}
dbGet().Must().Clauses(clause.OnConflict{
DoNothing: true,
}).Create(branches)
e2p(err)
return M{"message": "SUCCESS"}, nil
}
func Query(c *gin.Context) (interface{}, error) {
gid := c.Query("gid")
if gid == "" {

51
dtmsvr/dtmsvr_test.go
View File

@ -14,6 +14,9 @@ import (
"github.com/yedf/dtm/examples"
)
var DtmServer = examples.DtmServer
var Busi = examples.Busi
var myinit int = func() int {
common.InitApp(common.GetProjectDir(), &config)
config.Mysql["database"] = dbName
@ -50,7 +53,6 @@ func TestDtmSvr(t *testing.T) {
sagaNormal(t)
tccNormal(t)
tccRollback(t)
tccRollbackPending(t)
xaNormal(t)
xaRollback(t)
sagaCommittedPending(t)
@ -131,28 +133,27 @@ func xaRollback(t *testing.T) {
}
func tccNormal(t *testing.T) {
tcc := genTcc("gid-tcc-normal", false, false)
tcc.Submit()
assert.Equal(t, "submitted", getTransStatus(tcc.Gid))
WaitTransProcessed(tcc.Gid)
assert.Equal(t, []string{"prepared", "succeed", "succeed", "prepared", "succeed", "succeed"}, getBranchesStatus(tcc.Gid))
data := &examples.TransReq{Amount: 30}
_, err := dtmcli.TccGlobalTransaction(examples.DtmServer, func(tcc *dtmcli.Tcc) (rerr error) {
_, rerr = tcc.CallBranch(data, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert")
e2p(rerr)
_, rerr = tcc.CallBranch(data, Busi+"/TransIn", Busi+"/TransInConfirm", Busi+"/TransInRevert")
e2p(rerr)
return
})
e2p(err)
}
func tccRollback(t *testing.T) {
tcc := genTcc("gid-tcc-rollback", false, true)
tcc.Submit()
WaitTransProcessed(tcc.Gid)
assert.Equal(t, []string{"succeed", "prepared", "succeed", "succeed", "prepared", "failed"}, getBranchesStatus(tcc.Gid))
data := &examples.TransReq{Amount: 30, TransInResult: "FAIL"}
_, err := dtmcli.TccGlobalTransaction(examples.DtmServer, func(tcc *dtmcli.Tcc) (rerr error) {
_, rerr = tcc.CallBranch(data, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert")
e2p(rerr)
_, rerr = tcc.CallBranch(data, Busi+"/TransIn", Busi+"/TransInConfirm", Busi+"/TransInRevert")
e2p(rerr)
return
})
e2p(err)
}
func tccRollbackPending(t *testing.T) {
tcc := genTcc("gid-tcc-rollback-pending", false, true)
examples.MainSwitch.TransInRevertResult.SetOnce("PENDING")
tcc.Submit()
WaitTransProcessed(tcc.Gid)
// assert.Equal(t, "submitted", getTransStatus(tcc.Gid))
CronTransOnce(60*time.Second, "submitted")
assert.Equal(t, []string{"succeed", "prepared", "succeed", "succeed", "prepared", "failed"}, getBranchesStatus(tcc.Gid))
}
func msgNormal(t *testing.T) {
msg := genMsg("gid-normal-msg")
msg.Submit()
@ -226,16 +227,6 @@ func genSaga(gid string, outFailed bool, inFailed bool) *dtmcli.Saga {
return saga
}
func genTcc(gid string, outFailed bool, inFailed bool) *dtmcli.Tcc {
logrus.Printf("beginning a tcc test ---------------- %s", gid)
tcc := dtmcli.NewTcc(examples.DtmServer)
req := examples.GenTransReq(30, outFailed, inFailed)
tcc.Add(examples.Busi+"/TransOut", examples.Busi+"/TransOutConfirm", examples.Busi+"/TransOutRevert", &req)
tcc.Add(examples.Busi+"/TransIn", examples.Busi+"/TransInConfirm", examples.Busi+"/TransInRevert", &req)
tcc.Gid = gid
return tcc
}
func transQuery(t *testing.T, gid string) {
resp, err := common.RestyClient.R().SetQueryParam("gid", gid).Get(examples.DtmServer + "/query")
e2p(err)

38
examples/main_tcc.go
View File

@ -5,6 +5,7 @@ import (
"github.com/gin-gonic/gin"
"github.com/sirupsen/logrus"
"github.com/yedf/dtm/common"
"github.com/yedf/dtm/dtmcli"
)
@ -19,19 +20,36 @@ func TccMain() {
}
func TccSetup(app *gin.Engine) {
app.POST(BusiApi+"/TransInTcc", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
tcc, err := dtmcli.TccFromReq(c)
if err != nil {
return nil, err
}
req := reqFrom(c)
logrus.Printf("Trans in %f here, and Trans in another %f in call2 ", req.Amount/2, req.Amount/2)
_, rerr := tcc.CallBranch(&TransReq{Amount: req.Amount / 2}, Busi+"/TransIn", Busi+"/TransInConfirm", Busi+"/TransInRevert")
if rerr != nil {
return nil, rerr
}
return M{"result": "SUCCESS"}, nil
}))
}
func TccFireRequest() {
logrus.Printf("tcc transaction begin")
req := &TransReq{
Amount: 30,
TransInResult: "SUCCESS",
TransOutResult: "SUCCESS",
}
tcc := dtmcli.NewTcc(DtmServer).
Add(Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert", req).
Add(Busi+"/TransIn", Busi+"/TransInConfirm", Busi+"/TransOutRevert", req)
logrus.Printf("tcc trans submit")
err := tcc.Submit()
_, err := dtmcli.TccGlobalTransaction(DtmServer, func(tcc *dtmcli.Tcc) (rerr error) {
res1, rerr := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TransOut", Busi+"/TransOutConfirm", Busi+"/TransOutRevert")
if rerr != nil {
return
}
res2, rerr := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TransInTcc", Busi+"/TransInConfirm", Busi+"/TransInRevert")
if rerr != nil {
return
}
logrus.Printf("tcc returns: %s, %s", res1.String(), res2.String())
return
})
e2p(err)
}