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

xa add duplicate example and refactored

Browse Source
This commit is contained in:
yedf2 2021-09-03 19:43:09 +08:00
parent b5e0149dd4
commit 8cbab29d41
5 changed files with 154 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
dtmcli/types.go
View File

@ -37,6 +37,11 @@ func (g *IDGenerator) NewBranchID() string {
panic(fmt.Errorf("total branch id is longer than 20"))
}
g.branchID = g.branchID + 1
return g.CurrentBranchID()
}
// CurrentBranchID return current branchID
func (g *IDGenerator) CurrentBranchID() string {
return g.parentID + fmt.Sprintf("%02d", g.branchID)
}

73
dtmcli/xa.go
View File

@ -19,9 +19,7 @@ type XaRegisterCallback func(path string, xa *XaClient)
// XaClient xa client
type XaClient struct {
Server string
Conf map[string]string
NotifyURL string
XaClientBase
}
// Xa xa transaction
@ -40,11 +38,11 @@ func XaFromQuery(qs url.Values) (*Xa, error) {
// NewXaClient construct a xa client
func NewXaClient(server string, mysqlConf map[string]string, notifyURL string, register XaRegisterCallback) (*XaClient, error) {
xa := &XaClient{
xa := &XaClient{XaClientBase: XaClientBase{
Server: server,
Conf: mysqlConf,
NotifyURL: notifyURL,
}
}}
u, err := url.Parse(notifyURL)
if err != nil {
return nil, err
@ -55,15 +53,7 @@ func NewXaClient(server string, mysqlConf map[string]string, notifyURL string, r
// HandleCallback 处理commit/rollback的回调
func (xc *XaClient) HandleCallback(gid string, branchID string, action string) (interface{}, error) {
db, err := SdbAlone(xc.Conf)
if err != nil {
return nil, err
}
defer db.Close()
xaID := gid + "-" + branchID
_, err = DBExec(db, fmt.Sprintf("xa %s '%s'", action, xaID))
return ResultSuccess, err
return ResultSuccess, xc.XaClientBase.HandleCallback(gid, branchID, action)
}
// XaLocalTransaction start a xa local transaction
@ -72,62 +62,27 @@ func (xc *XaClient) XaLocalTransaction(qs url.Values, xaFunc XaLocalFunc) (ret i
if rerr != nil {
return
}
xa.Dtm = xc.Server
branchID := xa.NewBranchID()
xaBranch := xa.Gid + "-" + branchID
db, rerr := SdbAlone(xc.Conf)
if rerr != nil {
return
}
defer func() { db.Close() }()
defer func() {
x := recover()
_, err := DBExec(db, fmt.Sprintf("XA end '%s'", xaBranch))
if x == nil && rerr == nil && err == nil {
_, err = DBExec(db, fmt.Sprintf("XA prepare '%s'", xaBranch))
}
if rerr == nil {
rerr = err
}
if x != nil {
panic(x)
}
}()
_, rerr = DBExec(db, fmt.Sprintf("XA start '%s'", xaBranch))
if rerr != nil {
return
}
ret, rerr = xc.HandleLocalTrans(&xa.TransBase, func(db *sql.DB) (ret interface{}, rerr error) {
ret, rerr = xaFunc(db, xa)
rerr = CheckResult(ret, rerr)
if rerr != nil {
return
}
rerr = xa.callDtm(&M{"gid": xa.Gid, "branch_id": branchID, "trans_type": "xa", "url": xc.NotifyURL}, "registerXaBranch")
rerr = xa.callDtm(&M{"gid": xa.Gid, "branch_id": xa.CurrentBranchID(), "trans_type": "xa", "url": xc.XaClientBase.NotifyURL}, "registerXaBranch")
return
})
return
}
// XaGlobalTransaction start a xa global transaction
func (xc *XaClient) XaGlobalTransaction(gid string, xaFunc XaGlobalFunc) (rerr error) {
xa := Xa{TransBase: *NewTransBase(gid, "xa", xc.Server, "")}
rerr = xa.callDtm(xa, "prepare")
if rerr != nil {
return
}
// 小概率情况下，prepare成功了，但是由于网络状况导致上面Failure，那么不执行下面defer的内容，等待超时后再回滚标记事务失败，也没有问题
defer func() {
x := recover()
operation := If(x != nil || rerr != nil, "abort", "submit").(string)
err := xa.callDtm(xa, operation)
if rerr == nil { // 如果用户函数没有返回错误，那么返回dtm的
rerr = err
}
if x != nil {
panic(x)
}
}()
xa := Xa{TransBase: *NewTransBase(gid, "xa", xc.XaClientBase.Server, "")}
return xc.HandleGlobalTrans(&xa.TransBase, func(action string) error {
return xa.callDtm(xa, action)
}, func() error {
resp, rerr := xaFunc(&xa)
rerr = CheckResponse(resp, rerr)
return
return CheckResponse(resp, rerr)
})
}
// CallBranch call a xa branch

81
dtmcli/xa_base.go Normal file
View File

@ -0,0 +1,81 @@
package dtmcli
import (
"database/sql"
"fmt"
"strings"
)
// XaClientBase XaClient/XaGrpcClient base
type XaClientBase struct {
Server string
Conf map[string]string
NotifyURL string
}
// HandleCallback 处理commit/rollback的回调
func (xc *XaClientBase) HandleCallback(gid string, branchID string, action string) error {
db, err := SdbAlone(xc.Conf)
if err != nil {
return err
}
defer db.Close()
xaID := gid + "-" + branchID
_, err = DBExec(db, fmt.Sprintf("xa %s '%s'", action, xaID))
if err != nil && strings.Contains(err.Error(), "Error 1397: XAER_NOTA") { // 重复commit/rollback同一个id，报这个错误，忽略
err = nil
}
return err
}
// HandleLocalTrans http/grpc 处理LocalTransaction的公共方法
func (xc *XaClientBase) HandleLocalTrans(xa *TransBase, cb func(*sql.DB) (interface{}, error)) (ret interface{}, rerr error) {
branchID := xa.NewBranchID()
xaBranch := xa.Gid + "-" + branchID
db, rerr := SdbAlone(xc.Conf)
if rerr != nil {
return
}
defer func() { db.Close() }()
defer func() {
x := recover()
_, err := DBExec(db, fmt.Sprintf("XA end '%s'", xaBranch))
if x == nil && rerr == nil && err == nil {
_, err = DBExec(db, fmt.Sprintf("XA prepare '%s'", xaBranch))
}
if rerr == nil {
rerr = err
}
if x != nil {
panic(x)
}
}()
_, rerr = DBExec(db, fmt.Sprintf("XA start '%s'", xaBranch))
if rerr != nil {
return
}
ret, rerr = cb(db)
return
}
// HandleGlobalTrans http/grpc GlobalTransaction的公共方法
func (xc *XaClientBase) HandleGlobalTrans(xa *TransBase, callDtm func(string) error, callBusi func() error) (rerr error) {
rerr = callDtm("prepare")
if rerr != nil {
return
}
// 小概率情况下，prepare成功了，但是由于网络状况导致上面Failure，那么不执行下面defer的内容，等待超时后再回滚标记事务失败，也没有问题
defer func() {
x := recover()
operation := If(x != nil || rerr != nil, "abort", "submit").(string)
err := callDtm(operation)
if rerr == nil { // 如果用户函数没有返回错误，那么返回dtm的
rerr = err
}
if x != nil {
panic(x)
}
}()
rerr = callBusi()
return
}

86
dtmgrpc/xa.go
View File

@ -6,6 +6,8 @@ import (
"fmt"
"github.com/yedf/dtm/dtmcli"
grpc "google.golang.org/grpc"
emptypb "google.golang.org/protobuf/types/known/emptypb"
)
// XaGrpcGlobalFunc type of xa global function
@ -16,9 +18,7 @@ type XaGrpcLocalFunc func(db *sql.DB, xa *XaGrpc) error
// XaGrpcClient xa client
type XaGrpcClient struct {
Server string
Conf map[string]string
NotifyURL string
dtmcli.XaClientBase
}
// XaGrpc xa transaction
@ -39,25 +39,17 @@ func XaGrpcFromRequest(br *BusiRequest) (*XaGrpc, error) {
// NewXaGrpcClient construct a xa client
func NewXaGrpcClient(server string, mysqlConf map[string]string, notifyURL string) *XaGrpcClient {
xa := &XaGrpcClient{
xa := &XaGrpcClient{XaClientBase: dtmcli.XaClientBase{
Server: server,
Conf: mysqlConf,
NotifyURL: notifyURL,
}
}}
return xa
}
// HandleCallback 处理commit/rollback的回调
func (xc *XaGrpcClient) HandleCallback(gid string, branchID string, action string) error {
db, err := dtmcli.SdbAlone(xc.Conf)
if err != nil {
return err
}
defer db.Close()
xaID := gid + "-" + branchID
_, err = dtmcli.DBExec(db, fmt.Sprintf("xa %s '%s'", action, xaID))
return err
return xc.XaClientBase.HandleCallback(gid, branchID, action)
}
// XaLocalTransaction start a xa local transaction
@ -66,44 +58,22 @@ func (xc *XaGrpcClient) XaLocalTransaction(br *BusiRequest, xaFunc XaGrpcLocalFu
if rerr != nil {
return
}
xa.Dtm = xc.Server
branchID := xa.NewBranchID()
xaBranch := xa.Gid + "-" + branchID
db, rerr := dtmcli.SdbAlone(xc.Conf)
_, rerr = xc.HandleLocalTrans(&xa.TransBase, func(db *sql.DB) (interface{}, error) {
rerr := xaFunc(db, xa)
if rerr != nil {
return
}
defer func() { db.Close() }()
defer func() {
x := recover()
_, err := dtmcli.DBExec(db, fmt.Sprintf("XA end '%s'", xaBranch))
if x == nil && rerr == nil && err == nil {
_, err = dtmcli.DBExec(db, fmt.Sprintf("XA prepare '%s'", xaBranch))
}
if rerr == nil {
rerr = err
}
if x != nil {
panic(x)
}
}()
_, rerr = dtmcli.DBExec(db, fmt.Sprintf("XA start '%s'", xaBranch))
if rerr != nil {
return
}
rerr = xaFunc(db, xa)
if rerr != nil {
return
return nil, rerr
}
_, rerr = MustGetDtmClient(xa.Dtm).RegisterXaBranch(context.Background(), &DtmXaBranchRequest{
Info: &BranchInfo{
Gid: xa.Gid,
BranchID: branchID,
BranchID: xa.CurrentBranchID(),
TransType: xa.TransType,
},
BusiData: "",
Notify: xc.NotifyURL,
})
return nil, rerr
})
return
}
@ -115,27 +85,17 @@ func (xc *XaGrpcClient) XaGlobalTransaction(gid string, xaFunc XaGrpcGlobalFunc)
Gid: gid,
TransType: xa.TransType,
}
_, rerr = dc.Prepare(context.Background(), req)
if rerr != nil {
return
}
// 小概率情况下，prepare成功了，但是由于网络状况导致上面Failure，那么不执行下面defer的内容，等待超时后再回滚标记事务失败，也没有问题
defer func() {
x := recover()
if x == nil && rerr == nil {
_, rerr = dc.Submit(context.Background(), req)
return
}
_, err := dc.Abort(context.Background(), req)
if rerr == nil { // 如果用户函数没有返回错误，那么返回dtm的
rerr = err
}
if x != nil {
panic(x)
}
}()
rerr = xaFunc(&xa)
return
return xc.HandleGlobalTrans(&xa.TransBase, func(action string) error {
f := map[string]func(context.Context, *DtmRequest, ...grpc.CallOption) (*emptypb.Empty, error){
"prepare": dc.Prepare,
"submit": dc.Submit,
"abort": dc.Abort,
}[action]
_, err := f(context.Background(), req)
return err
}, func() error {
return xaFunc(&xa)
})
}
// CallBranch call a xa branch

19
test/xa_test.go
View File

@ -6,6 +6,7 @@ import (
"github.com/go-resty/resty/v2"
"github.com/stretchr/testify/assert"
"github.com/yedf/dtm/common"
"github.com/yedf/dtm/dtmcli"
"github.com/yedf/dtm/examples"
)
@ -16,6 +17,7 @@ func TestXa(t *testing.T) {
}
xaLocalError(t)
xaNormal(t)
xaDuplicate(t)
xaRollback(t)
}
@ -43,6 +45,23 @@ func xaNormal(t *testing.T) {
assert.Equal(t, []string{"prepared", "succeed", "prepared", "succeed"}, getBranchesStatus(gid))
}
func xaDuplicate(t *testing.T) {
xc := examples.XaClient
gid := "xaDuplicate"
err := xc.XaGlobalTransaction(gid, func(xa *dtmcli.Xa) (*resty.Response, error) {
req := examples.GenTransReq(30, false, false)
_, err := xa.CallBranch(req, examples.Busi+"/TransOutXa")
assert.Nil(t, err)
sdb, err := dtmcli.SdbAlone(common.DtmConfig.DB)
assert.Nil(t, err)
dtmcli.DBExec(sdb, "xa recover")
dtmcli.DBExec(sdb, "xa commit 'xaDuplicate-0101'") // 先把某一个事务提交，模拟重复请求
return xa.CallBranch(req, examples.Busi+"/TransInXa")
})
assert.Equal(t, nil, err)
WaitTransProcessed(gid)
assert.Equal(t, []string{"prepared", "succeed", "prepared", "succeed"}, getBranchesStatus(gid))
}
func xaRollback(t *testing.T) {
xc := examples.XaClient
gid := "xaRollback"