grpc xa left a bug to fix

dtmgrpc/xa.go

@@ -1,25 +1,21 @@
 package dtmgrpc
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
-	"net/url"
 
-	"github.com/go-resty/resty/v2"
 	"github.com/yedf/dtm/dtmcli"
 )
 
-// XaGlobalFunc type of xa global function
+// XaGrpcGlobalFunc type of xa global function
-type XaGlobalFunc func(xa *XaGrpc) (*resty.Response, error)
+type XaGrpcGlobalFunc func(xa *XaGrpc) error
 
-// XaLocalFunc type of xa local function
+// XaGrpcLocalFunc type of xa local function
-type XaLocalFunc func(db *sql.DB, xa *XaGrpc) (interface{}, error)
+type XaGrpcLocalFunc func(db *sql.DB, xa *XaGrpc) error
 
-// XaRegisterCallback type of xa register callback handler
+// XaGrpcClient xa client
-type XaRegisterCallback func(path string, xa *XaClient)
+type XaGrpcClient struct {
-
-// XaClient xa client
-type XaClient struct {
 	Server    string
 	Conf      map[string]string
 	NotifyURL string
@@ -31,11 +27,11 @@ type XaGrpc struct {
 	dtmcli.TransBase
 }
 
-// XaFromRequest construct xa info from request
+// XaGrpcFromRequest construct xa info from request
-func XaFromRequest(br *BusiRequest) (*XaGrpc, error) {
+func XaGrpcFromRequest(br *BusiRequest) (*XaGrpc, error) {
 	xa := &XaGrpc{
 		TransBase: *dtmcli.NewTransBase(br.Dtm, br.Info.BranchID),
-		TransData: dtmcli.TransData{Gid: br.Info.BranchID, TransType: br.Info.TransType},
+		TransData: dtmcli.TransData{Gid: br.Info.Gid, TransType: br.Info.TransType},
 	}
 	if xa.Gid == "" || br.Info.BranchID == "" {
 		return nil, fmt.Errorf("bad xa info: gid: %s parentid: %s", xa.Gid, br.Info.BranchID)
@@ -43,37 +39,32 @@ func XaFromRequest(br *BusiRequest) (*XaGrpc, error) {
 	return xa, nil
 }
 
-// NewXaClient construct a xa client
+// NewXaGrpcClient construct a xa client
-func NewXaClient(server string, mysqlConf map[string]string, notifyURL string, register XaRegisterCallback) (*XaClient, error) {
+func NewXaGrpcClient(server string, mysqlConf map[string]string, notifyURL string) *XaGrpcClient {
-	xa := &XaClient{
+	xa := &XaGrpcClient{
 		Server:    server,
 		Conf:      mysqlConf,
 		NotifyURL: notifyURL,
 	}
-	u, err := url.Parse(notifyURL)
+	return xa
-	if err != nil {
-		return nil, err
-	}
-	register(u.Path, xa)
-	return xa, nil
 }
 
 // HandleCallback 处理commit/rollback的回调
-func (xc *XaClient) HandleCallback(gid string, branchID string, action string) (interface{}, error) {
+func (xc *XaGrpcClient) HandleCallback(gid string, branchID string, action string) error {
 	db, err := dtmcli.SdbAlone(xc.Conf)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	defer db.Close()
 	xaID := gid + "-" + branchID
 	_, err = dtmcli.SdbExec(db, fmt.Sprintf("xa %s '%s'", action, xaID))
-	return dtmcli.ResultSuccess, err
+	return err
 
 }
 
 // XaLocalTransaction start a xa local transaction
-func (xc *XaClient) XaLocalTransaction(br *BusiRequest, xaFunc XaLocalFunc) (ret interface{}, rerr error) {
+func (xc *XaGrpcClient) XaLocalTransaction(br *BusiRequest, xaFunc XaGrpcLocalFunc) (rerr error) {
-	xa, rerr := XaFromRequest(br)
+	xa, rerr := XaGrpcFromRequest(br)
 	if rerr != nil {
 		return
 	}
@@ -102,28 +93,42 @@ func (xc *XaClient) XaLocalTransaction(br *BusiRequest, xaFunc XaLocalFunc) (ret
 	if rerr != nil {
 		return
 	}
-	ret, rerr = xaFunc(db, xa)
+	rerr = xaFunc(db, xa)
-	rerr = dtmcli.CheckResult(ret, rerr)
 	if rerr != nil {
 		return
 	}
-	rerr = xa.CallDtm(&dtmcli.M{"gid": xa.Gid, "branch_id": branchID, "trans_type": "xa", "url": xc.NotifyURL}, "registerXaBranch")
+	_, rerr = MustGetDtmClient(xa.Dtm).RegisterXaBranch(context.Background(), &DtmXaBranchRequest{
+		Info: &BranchInfo{
+			Gid:       xa.Gid,
+			BranchID:  branchID,
+			TransType: xa.TransType,
+		},
+		BusiData: "",
+		Notify:   xc.NotifyURL,
+	})
 	return
 }
 
 // XaGlobalTransaction start a xa global transaction
-func (xc *XaClient) XaGlobalTransaction(gid string, xaFunc XaGlobalFunc) (rerr error) {
+func (xc *XaGrpcClient) XaGlobalTransaction(gid string, xaFunc XaGrpcGlobalFunc) (rerr error) {
-	xa := XaGrpc{TransBase: dtmcli.TransBase{IDGenerator: dtmcli.IDGenerator{}, Dtm: xc.Server}, TransData: dtmcli.TransData{Gid: gid, TransType: "xa"}}
+	xa := XaGrpc{TransBase: dtmcli.TransBase{Dtm: xc.Server}, TransData: dtmcli.TransData{Gid: gid, TransType: "xa"}}
-	rerr = xa.CallDtm(&xa.TransData, "prepare")
+	dc := MustGetDtmClient(xa.Dtm)
+	req := &DtmRequest{
+		Gid:       gid,
+		TransType: xa.TransType,
+	}
+	_, rerr = dc.Prepare(context.Background(), req)
 	if rerr != nil {
 		return
 	}
-	var resp *resty.Response
 	// 小概率情况下，prepare成功了，但是由于网络状况导致上面Failure，那么不执行下面defer的内容，等待超时后再回滚标记事务失败，也没有问题
 	defer func() {
 		x := recover()
-		operation := dtmcli.If(x != nil || rerr != nil, "abort", "submit").(string)
+		if x == nil && rerr == nil {
-		err := xa.CallDtm(&xa.TransData, operation)
+			_, rerr = dc.Submit(context.Background(), req)
+			return
+		}
+		_, err := dc.Abort(context.Background(), req)
 		if rerr == nil { // 如果用户函数没有返回错误，那么返回dtm的
 			rerr = err
 		}
@@ -131,22 +136,25 @@ func (xc *XaClient) XaGlobalTransaction(gid string, xaFunc XaGlobalFunc) (rerr e
 			panic(x)
 		}
 	}()
-	resp, rerr = xaFunc(&xa)
+	rerr = xaFunc(&xa)
-	rerr = dtmcli.CheckResponse(resp, rerr)
 	return
 }
 
 // CallBranch call a xa branch
-func (x *XaGrpc) CallBranch(body interface{}, url string) (*resty.Response, error) {
+func (x *XaGrpc) CallBranch(busiData []byte, url string) (*BusiReply, error) {
 	branchID := x.NewBranchID()
-	resp, err := dtmcli.RestyClient.R().
+	server, method := GetServerAndMethod(url)
-		SetBody(body).
+	reply := &BusiReply{}
-		SetQueryParams(dtmcli.MS{
+	err := MustGetGrpcConn(server).Invoke(context.Background(), method, &BusiRequest{
-			"gid":         x.Gid,
+		Info: &BranchInfo{
-			"branch_id":   branchID,
+			Gid:        x.Gid,
-			"trans_type":  "xa",
+			TransType:  x.TransType,
-			"branch_type": "action",
+			BranchID:   branchID,
-		}).
+			BranchType: "",
-		Post(url)
+		},
-	return resp, dtmcli.CheckResponse(resp, err)
+		Dtm:      x.Dtm,
+		BusiData: busiData,
+	}, reply)
+	return reply, err
+
 }

dtmsvr/api.go

@@ -54,3 +54,21 @@ func svcRegisterTccBranch(branch *TransBranch, data dtmcli.MS) (interface{}, err
 	global.touch(dbGet(), config.TransCronInterval)
 	return dtmcli.ResultSuccess, nil
 }
+
+func svcRegisterXaBranch(branch *TransBranch) (interface{}, error) {
+	branch.Status = "prepared"
+	db := dbGet()
+	dbt := TransFromDb(db, branch.Gid)
+	if dbt.Status != "prepared" {
+		return M{"dtm_result": "FAILURE", "message": fmt.Sprintf("current status: %s cannot register branch", dbt.Status)}, nil
+	}
+	branches := []TransBranch{*branch, *branch}
+	branches[0].BranchType = "rollback"
+	branches[1].BranchType = "commit"
+	db.Must().Clauses(clause.OnConflict{
+		DoNothing: true,
+	}).Create(branches)
+	global := TransGlobal{Gid: branch.Gid}
+	global.touch(db, config.TransCronInterval)
+	return dtmcli.ResultSuccess, nil
+}

dtmsvr/api_grpc.go

@@ -42,3 +42,13 @@ func (s *dtmServer) RegisterTccBranch(ctx context.Context, in *pb.DtmTccBranchRe
 	})
 	return &emptypb.Empty{}, dtmgrpc.Result2Error(r, err)
 }
+
+func (s *dtmServer) RegisterXaBranch(ctx context.Context, in *pb.DtmXaBranchRequest) (*emptypb.Empty, error) {
+	r, err := svcRegisterXaBranch(&TransBranch{
+		Gid:      in.Info.Gid,
+		BranchID: in.Info.BranchID,
+		Status:   "prepared",
+		Data:     in.BusiData,
+	})
+	return &emptypb.Empty{}, dtmgrpc.Result2Error(r, err)
+}

dtmsvr/api_http.go

@@ -2,13 +2,11 @@ package dtmsvr
 
 import (
 	"errors"
-	"fmt"
 
 	"github.com/gin-gonic/gin"
 	"github.com/yedf/dtm/common"
 	"github.com/yedf/dtm/dtmcli"
 	"gorm.io/gorm"
-	"gorm.io/gorm/clause"
 )
 
 func addRoute(engine *gin.Engine) {
@@ -44,22 +42,7 @@ func registerXaBranch(c *gin.Context) (interface{}, error) {
 	branch := TransBranch{}
 	err := c.BindJSON(&branch)
 	e2p(err)
-	branch.Status = "prepared"
+	return svcRegisterXaBranch(&branch)
-	db := dbGet()
-	dbt := TransFromDb(db, branch.Gid)
-	if dbt.Status != "prepared" {
-		return M{"dtm_result": "FAILURE", "message": fmt.Sprintf("current status: %s cannot register branch", dbt.Status)}, nil
-	}
-	branches := []TransBranch{branch, branch}
-	branches[0].BranchType = "rollback"
-	branches[1].BranchType = "commit"
-	db.Must().Clauses(clause.OnConflict{
-		DoNothing: true,
-	}).Create(branches)
-	e2p(err)
-	global := TransGlobal{Gid: branch.Gid}
-	global.touch(db, config.TransCronInterval)
-	return dtmcli.ResultSuccess, nil
 }
 
 func registerTccBranch(c *gin.Context) (interface{}, error) {

dtmsvr/dtmsvr_test.go

@@ -48,10 +48,9 @@ func TestMain(m *testing.M) {
 	go StartSvr()
 	app = examples.BaseAppStartup()
 	examples.GrpcStartup()
-	examples.SagaSetup(app)
+	examples.XaGrpcSetup()
 	examples.TccSetup(app)
 	examples.XaSetup(app)
-	examples.MsgSetup(app)
 	examples.TccBarrierAddRoute(app)
 	examples.SagaBarrierAddRoute(app)
 

dtmsvr/trans_grpc_xa_test.go

@@ -0,0 +1,63 @@
+package dtmsvr
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/yedf/dtm/dtmcli"
+	"github.com/yedf/dtm/dtmgrpc"
+	"github.com/yedf/dtm/examples"
+)
+
+func TestGrpcXa(t *testing.T) {
+	if config.DB["driver"] != "mysql" {
+		return
+	}
+	// xaGrpcLocalError(t)
+	xaGrpcNormal(t)
+	xaGrpcRollback(t)
+}
+
+func xaGrpcLocalError(t *testing.T) {
+	xc := examples.XaGrpcClient
+	err := xc.XaGlobalTransaction("xaGrpcLocalError", func(xa *dtmgrpc.XaGrpc) error {
+		return fmt.Errorf("an error")
+	})
+	assert.Error(t, err, fmt.Errorf("an error"))
+}
+
+func xaGrpcNormal(t *testing.T) {
+	xc := examples.XaGrpcClient
+	gid := "xaGrpcNormal"
+	err := xc.XaGlobalTransaction(gid, func(xa *dtmgrpc.XaGrpc) error {
+		req := dtmcli.MustMarshal(examples.GenTransReq(30, false, false))
+		_, err := xa.CallBranch(req, examples.BusiGrpc+"/examples.Busi/TransOutXa")
+		if err != nil {
+			return err
+		}
+		_, err = xa.CallBranch(req, examples.BusiGrpc+"/examples.Busi/TransInXa")
+		return err
+	})
+	assert.Equal(t, nil, err)
+	WaitTransProcessed(gid)
+	assert.Equal(t, []string{"prepared", "succeed", "prepared", "succeed"}, getBranchesStatus(gid))
+}
+
+func xaGrpcRollback(t *testing.T) {
+	xc := examples.XaGrpcClient
+	gid := "xaGrpcRollback"
+	err := xc.XaGlobalTransaction(gid, func(xa *dtmgrpc.XaGrpc) error {
+		req := dtmcli.MustMarshal(&examples.TransReq{Amount: 30, TransInResult: "FAILURE"})
+		_, err := xa.CallBranch(req, examples.BusiGrpc+"/examples.Busi/TransOutXa")
+		if err != nil {
+			return err
+		}
+		_, err = xa.CallBranch(req, examples.BusiGrpc+"/examples.Busi/TransInXa")
+		return err
+	})
+	assert.Error(t, err)
+	WaitTransProcessed(gid)
+	assert.Equal(t, []string{"succeed", "prepared"}, getBranchesStatus(gid))
+	assert.Equal(t, "failed", getTransStatus(gid))
+}

examples/busi.pb.go

@@ -29,7 +29,7 @@ var file_examples_busi_proto_rawDesc = []byte{
 	0x15, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x2f, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63,
 	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1b, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70,
 	0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x65, 0x6d, 0x70, 0x74, 0x79, 0x2e, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x32, 0xc2, 0x03, 0x0a, 0x04, 0x42, 0x75, 0x73, 0x69, 0x12, 0x3b, 0x0a, 0x09,
+	0x6f, 0x74, 0x6f, 0x32, 0xf9, 0x04, 0x0a, 0x04, 0x42, 0x75, 0x73, 0x69, 0x12, 0x3b, 0x0a, 0x09,
 	0x43, 0x61, 0x6e, 0x53, 0x75, 0x62, 0x6d, 0x69, 0x74, 0x12, 0x14, 0x2e, 0x64, 0x74, 0x6d, 0x67,
 	0x72, 0x70, 0x63, 0x2e, 0x42, 0x75, 0x73, 0x69, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
 	0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75,
@@ -57,9 +57,21 @@ var file_examples_busi_proto_rawDesc = []byte{
 	0x74, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x72, 0x6d, 0x12, 0x14, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x72,
 	0x70, 0x63, 0x2e, 0x42, 0x75, 0x73, 0x69, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16,
 	0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66,
-	0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x42, 0x1e, 0x5a, 0x1c, 0x67, 0x69, 0x74, 0x68,
+	0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x3a, 0x0a, 0x08, 0x58, 0x61, 0x4e, 0x6f,
-	0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x79, 0x65, 0x64, 0x66, 0x2f, 0x64, 0x74, 0x6d, 0x2f,
+	0x74, 0x69, 0x66, 0x79, 0x12, 0x14, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x42,
-	0x65, 0x78, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x75, 0x73, 0x69, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f,
+	0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70,
+	0x74, 0x79, 0x22, 0x00, 0x12, 0x3b, 0x0a, 0x09, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x49, 0x6e, 0x58,
+	0x61, 0x12, 0x14, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x42, 0x75, 0x73, 0x69,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65,
+	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22,
+	0x00, 0x12, 0x3c, 0x0a, 0x0a, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x4f, 0x75, 0x74, 0x58, 0x61, 0x12,
+	0x14, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x42, 0x75, 0x73, 0x69, 0x52, 0x65,
+	0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x42,
+	0x1e, 0x5a, 0x1c, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x79, 0x65,
+	0x64, 0x66, 0x2f, 0x64, 0x74, 0x6d, 0x2f, 0x65, 0x78, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x73, 0x62,
+	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var file_examples_busi_proto_goTypes = []interface{}{
@@ -74,15 +86,21 @@ var file_examples_busi_proto_depIdxs = []int32{
 	0,  // 4: examples.Busi.TransOutRevert:input_type -> dtmgrpc.BusiRequest
 	0,  // 5: examples.Busi.TransInConfirm:input_type -> dtmgrpc.BusiRequest
 	0,  // 6: examples.Busi.TransOutConfirm:input_type -> dtmgrpc.BusiRequest
-	1, // 7: examples.Busi.CanSubmit:output_type -> google.protobuf.Empty
+	0,  // 7: examples.Busi.XaNotify:input_type -> dtmgrpc.BusiRequest
-	1, // 8: examples.Busi.TransIn:output_type -> google.protobuf.Empty
+	0,  // 8: examples.Busi.TransInXa:input_type -> dtmgrpc.BusiRequest
-	1, // 9: examples.Busi.TransOut:output_type -> google.protobuf.Empty
+	0,  // 9: examples.Busi.TransOutXa:input_type -> dtmgrpc.BusiRequest
-	1, // 10: examples.Busi.TransInRevert:output_type -> google.protobuf.Empty
+	1,  // 10: examples.Busi.CanSubmit:output_type -> google.protobuf.Empty
-	1, // 11: examples.Busi.TransOutRevert:output_type -> google.protobuf.Empty
+	1,  // 11: examples.Busi.TransIn:output_type -> google.protobuf.Empty
-	1, // 12: examples.Busi.TransInConfirm:output_type -> google.protobuf.Empty
+	1,  // 12: examples.Busi.TransOut:output_type -> google.protobuf.Empty
-	1, // 13: examples.Busi.TransOutConfirm:output_type -> google.protobuf.Empty
+	1,  // 13: examples.Busi.TransInRevert:output_type -> google.protobuf.Empty
-	7, // [7:14] is the sub-list for method output_type
+	1,  // 14: examples.Busi.TransOutRevert:output_type -> google.protobuf.Empty
-	0, // [0:7] is the sub-list for method input_type
+	1,  // 15: examples.Busi.TransInConfirm:output_type -> google.protobuf.Empty
+	1,  // 16: examples.Busi.TransOutConfirm:output_type -> google.protobuf.Empty
+	1,  // 17: examples.Busi.XaNotify:output_type -> google.protobuf.Empty
+	1,  // 18: examples.Busi.TransInXa:output_type -> google.protobuf.Empty
+	1,  // 19: examples.Busi.TransOutXa:output_type -> google.protobuf.Empty
+	10, // [10:20] is the sub-list for method output_type
+	0,  // [0:10] is the sub-list for method input_type
 	0,  // [0:0] is the sub-list for extension type_name
 	0,  // [0:0] is the sub-list for extension extendee
 	0,  // [0:0] is the sub-list for field type_name

examples/busi.proto

@@ -15,5 +15,8 @@ service Busi {
   rpc TransOutRevert(dtmgrpc.BusiRequest) returns (google.protobuf.Empty) {}
   rpc TransInConfirm(dtmgrpc.BusiRequest) returns (google.protobuf.Empty) {}
   rpc TransOutConfirm(dtmgrpc.BusiRequest) returns (google.protobuf.Empty) {}
+  rpc XaNotify(dtmgrpc.BusiRequest) returns (google.protobuf.Empty) {}
+  rpc TransInXa(dtmgrpc.BusiRequest) returns (google.protobuf.Empty) {}
+  rpc TransOutXa(dtmgrpc.BusiRequest) returns (google.protobuf.Empty) {}
 }
 

examples/busi_grpc.pb.go

@@ -27,6 +27,9 @@ type BusiClient interface {
 	TransOutRevert(ctx context.Context, in *dtmgrpc.BusiRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
 	TransInConfirm(ctx context.Context, in *dtmgrpc.BusiRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
 	TransOutConfirm(ctx context.Context, in *dtmgrpc.BusiRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
+	XaNotify(ctx context.Context, in *dtmgrpc.BusiRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
+	TransInXa(ctx context.Context, in *dtmgrpc.BusiRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
+	TransOutXa(ctx context.Context, in *dtmgrpc.BusiRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
 }
 
 type busiClient struct {
@@ -100,6 +103,33 @@ func (c *busiClient) TransOutConfirm(ctx context.Context, in *dtmgrpc.BusiReques
 	return out, nil
 }
 
+func (c *busiClient) XaNotify(ctx context.Context, in *dtmgrpc.BusiRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+	out := new(emptypb.Empty)
+	err := c.cc.Invoke(ctx, "/examples.Busi/XaNotify", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *busiClient) TransInXa(ctx context.Context, in *dtmgrpc.BusiRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+	out := new(emptypb.Empty)
+	err := c.cc.Invoke(ctx, "/examples.Busi/TransInXa", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *busiClient) TransOutXa(ctx context.Context, in *dtmgrpc.BusiRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+	out := new(emptypb.Empty)
+	err := c.cc.Invoke(ctx, "/examples.Busi/TransOutXa", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // BusiServer is the server API for Busi service.
 // All implementations must embed UnimplementedBusiServer
 // for forward compatibility
@@ -111,6 +141,9 @@ type BusiServer interface {
 	TransOutRevert(context.Context, *dtmgrpc.BusiRequest) (*emptypb.Empty, error)
 	TransInConfirm(context.Context, *dtmgrpc.BusiRequest) (*emptypb.Empty, error)
 	TransOutConfirm(context.Context, *dtmgrpc.BusiRequest) (*emptypb.Empty, error)
+	XaNotify(context.Context, *dtmgrpc.BusiRequest) (*emptypb.Empty, error)
+	TransInXa(context.Context, *dtmgrpc.BusiRequest) (*emptypb.Empty, error)
+	TransOutXa(context.Context, *dtmgrpc.BusiRequest) (*emptypb.Empty, error)
 	mustEmbedUnimplementedBusiServer()
 }
 
@@ -139,6 +172,15 @@ func (UnimplementedBusiServer) TransInConfirm(context.Context, *dtmgrpc.BusiRequ
 func (UnimplementedBusiServer) TransOutConfirm(context.Context, *dtmgrpc.BusiRequest) (*emptypb.Empty, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method TransOutConfirm not implemented")
 }
+func (UnimplementedBusiServer) XaNotify(context.Context, *dtmgrpc.BusiRequest) (*emptypb.Empty, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method XaNotify not implemented")
+}
+func (UnimplementedBusiServer) TransInXa(context.Context, *dtmgrpc.BusiRequest) (*emptypb.Empty, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method TransInXa not implemented")
+}
+func (UnimplementedBusiServer) TransOutXa(context.Context, *dtmgrpc.BusiRequest) (*emptypb.Empty, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method TransOutXa not implemented")
+}
 func (UnimplementedBusiServer) mustEmbedUnimplementedBusiServer() {}
 
 // UnsafeBusiServer may be embedded to opt out of forward compatibility for this service.
@@ -278,6 +320,60 @@ func _Busi_TransOutConfirm_Handler(srv interface{}, ctx context.Context, dec fun
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Busi_XaNotify_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(dtmgrpc.BusiRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(BusiServer).XaNotify(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/examples.Busi/XaNotify",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(BusiServer).XaNotify(ctx, req.(*dtmgrpc.BusiRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Busi_TransInXa_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(dtmgrpc.BusiRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(BusiServer).TransInXa(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/examples.Busi/TransInXa",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(BusiServer).TransInXa(ctx, req.(*dtmgrpc.BusiRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Busi_TransOutXa_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(dtmgrpc.BusiRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(BusiServer).TransOutXa(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/examples.Busi/TransOutXa",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(BusiServer).TransOutXa(ctx, req.(*dtmgrpc.BusiRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // Busi_ServiceDesc is the grpc.ServiceDesc for Busi service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -313,6 +409,18 @@ var Busi_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "TransOutConfirm",
 			Handler:    _Busi_TransOutConfirm_Handler,
 		},
+		{
+			MethodName: "XaNotify",
+			Handler:    _Busi_XaNotify_Handler,
+		},
+		{
+			MethodName: "TransInXa",
+			Handler:    _Busi_TransInXa_Handler,
+		},
+		{
+			MethodName: "TransOutXa",
+			Handler:    _Busi_TransOutXa_Handler,
+		},
 	},
 	Streams:  []grpc.StreamDesc{},
 	Metadata: "examples/busi.proto",

examples/main_base.go

@@ -1,6 +1,7 @@
 package examples
 
 import (
+	"database/sql"
 	"fmt"
 	"time"
 
@@ -18,6 +19,10 @@ const (
 	BusiGrpcPort = 60081
 )
 
+type setupFunc func(*gin.Engine)
+
+var setupFuncs = map[string]setupFunc{}
+
 // Busi busi service url prefix
 var Busi string = fmt.Sprintf("http://localhost:%d%s", BusiPort, BusiAPI)
 
@@ -26,6 +31,10 @@ func BaseAppStartup() *gin.Engine {
 	dtmcli.Logf("examples starting")
 	app := common.GetGinApp()
 	BaseAddRoute(app)
+	for k, v := range setupFuncs {
+		dtmcli.Logf("initing %s", k)
+		v(app)
+	}
 	dtmcli.Logf("Starting busi at: %d", BusiPort)
 	go app.Run(fmt.Sprintf(":%d", BusiPort))
 
@@ -94,4 +103,23 @@ func BaseAddRoute(app *gin.Engine) {
 		dtmcli.Logf("%s CanSubmit", c.Query("gid"))
 		return dtmcli.OrString(MainSwitch.CanSubmitResult.Fetch(), "SUCCESS"), nil
 	}))
+	app.POST(BusiAPI+"/TransInXa", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
+		return XaClient.XaLocalTransaction(c.Request.URL.Query(), func(db *sql.DB, xa *dtmcli.Xa) (interface{}, error) {
+			if reqFrom(c).TransInResult == "FAILURE" {
+				return dtmcli.ResultFailure, nil
+			}
+			_, err := dtmcli.SdbExec(db, "update dtm_busi.user_account set balance=balance+? where user_id=?", reqFrom(c).Amount, 2)
+			return dtmcli.ResultSuccess, err
+		})
+	}))
+	app.POST(BusiAPI+"/TransOutXa", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
+		return XaClient.XaLocalTransaction(c.Request.URL.Query(), func(db *sql.DB, xa *dtmcli.Xa) (interface{}, error) {
+			if reqFrom(c).TransOutResult == "FAILURE" {
+				return dtmcli.ResultFailure, nil
+			}
+			_, err := dtmcli.SdbExec(db, "update dtm_busi.user_account set balance=balance-? where user_id=?", reqFrom(c).Amount, 1)
+			return dtmcli.ResultSuccess, err
+		})
+	}))
+
 }

examples/main_grpc.go

@@ -2,6 +2,7 @@ package examples
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"net"
 
@@ -94,3 +95,27 @@ func (s *busiServer) TransOutConfirm(ctx context.Context, in *dtmgrpc.BusiReques
 	dtmcli.MustUnmarshal(in.BusiData, &req)
 	return &emptypb.Empty{}, handleGrpcBusiness(in, MainSwitch.TransOutConfirmResult.Fetch(), "", dtmcli.GetFuncName())
 }
+
+func (s *busiServer) TransInXa(ctx context.Context, in *dtmgrpc.BusiRequest) (*emptypb.Empty, error) {
+	req := TransReq{}
+	dtmcli.MustUnmarshal(in.BusiData, &req)
+	return &emptypb.Empty{}, XaGrpcClient.XaLocalTransaction(in, func(db *sql.DB, xa *dtmgrpc.XaGrpc) error {
+		if req.TransInResult == "FAILURE" {
+			return status.New(codes.Aborted, "user return failure").Err()
+		}
+		_, err := dtmcli.SdbExec(db, "update dtm_busi.user_account set balance=balance+? where user_id=?", req.Amount, 2)
+		return err
+	})
+}
+
+func (s *busiServer) TransOutXa(ctx context.Context, in *dtmgrpc.BusiRequest) (*emptypb.Empty, error) {
+	req := TransReq{}
+	dtmcli.MustUnmarshal(in.BusiData, &req)
+	return &emptypb.Empty{}, XaGrpcClient.XaLocalTransaction(in, func(db *sql.DB, xa *dtmgrpc.XaGrpc) error {
+		if req.TransOutResult == "FAILURE" {
+			return status.New(codes.Aborted, "user return failure").Err()
+		}
+		_, err := dtmcli.SdbExec(db, "update dtm_busi.user_account set balance=balance-? where user_id=?", req.Amount, 1)
+		return err
+	})
+}

examples/main_grpc_msg.go

@@ -1,16 +1,10 @@
 package examples
 
 import (
-	"github.com/gin-gonic/gin"
 	"github.com/yedf/dtm/dtmcli"
 	dtmgrpc "github.com/yedf/dtm/dtmgrpc"
 )
 
-// MsgGrpcSetup 1
-func MsgGrpcSetup(app *gin.Engine) {
-
-}
-
 // MsgGrpcFireRequest 1
 func MsgGrpcFireRequest() string {
 	req := dtmcli.MustMarshal(&TransReq{Amount: 30})

examples/main_grpc_xa.go

@@ -0,0 +1,38 @@
+package examples
+
+import (
+	context "context"
+
+	"github.com/yedf/dtm/dtmcli"
+	"github.com/yedf/dtm/dtmgrpc"
+	emptypb "google.golang.org/protobuf/types/known/emptypb"
+)
+
+// XaGrpcClient XA client connection
+var XaGrpcClient *dtmgrpc.XaGrpcClient = nil
+
+// XaGrpcSetup 挂载http的api，创建XaClient
+func XaGrpcSetup() {
+	XaGrpcClient = dtmgrpc.NewXaGrpcClient(DtmGrpcServer, config.DB, BusiGrpc+"/examples.Busi/XaNotify")
+}
+
+func (s *busiServer) XaNotify(ctx context.Context, in *dtmgrpc.BusiRequest) (*emptypb.Empty, error) {
+	err := XaGrpcClient.HandleCallback(in.Info.Gid, in.Info.BranchID, in.Info.BranchType)
+	return &emptypb.Empty{}, dtmgrpc.Result2Error(nil, err)
+}
+
+// XaGrpcFireRequest 注册全局XA事务，调用XA的分支
+func XaGrpcFireRequest() string {
+	gid := dtmcli.MustGenGid(DtmServer)
+	busiData := dtmcli.MustMarshal(&TransReq{Amount: 30})
+	err := XaGrpcClient.XaGlobalTransaction(gid, func(xa *dtmgrpc.XaGrpc) error {
+		_, err := xa.CallBranch(busiData, BusiGrpc+"/examples.Busi/TransOutXa")
+		if err != nil {
+			return err
+		}
+		_, err = xa.CallBranch(busiData, BusiGrpc+"/examples.Busi/TransInXa")
+		return err
+	})
+	e2p(err)
+	return gid
+}

examples/main_msg.go

@@ -1,14 +1,9 @@
 package examples
 
 import (
-	"github.com/gin-gonic/gin"
 	"github.com/yedf/dtm/dtmcli"
 )
 
-// MsgSetup 1
-func MsgSetup(app *gin.Engine) {
-}
-
 // MsgFireRequest 1
 func MsgFireRequest() string {
 	dtmcli.Logf("a busi transaction begin")

examples/main_saga.go

@@ -1,14 +1,9 @@
 package examples
 
 import (
-	"github.com/gin-gonic/gin"
 	"github.com/yedf/dtm/dtmcli"
 )
 
-// SagaSetup 1
-func SagaSetup(app *gin.Engine) {
-}
-
 // SagaFireRequest 1
 func SagaFireRequest() string {
 	dtmcli.Logf("a saga busi transaction begin")

examples/main_saga_wait.go

@@ -1,14 +1,9 @@
 package examples
 
 import (
-	"github.com/gin-gonic/gin"
 	"github.com/yedf/dtm/dtmcli"
 )
 
-// SagaWaitSetup 1
-func SagaWaitSetup(app *gin.Engine) {
-}
-
 // SagaWaitFireRequest 1
 func SagaWaitFireRequest() string {
 	dtmcli.Logf("a saga busi transaction begin")

examples/main_xa.go

@@ -1,8 +1,6 @@
 package examples
 
 import (
-	"database/sql"
-
 	"github.com/gin-gonic/gin"
 	"github.com/go-resty/resty/v2"
 	"github.com/yedf/dtm/common"
@@ -14,8 +12,6 @@ var XaClient *dtmcli.XaClient = nil
 
 // XaSetup 挂载http的api，创建XaClient
 func XaSetup(app *gin.Engine) {
-	app.POST(BusiAPI+"/TransInXa", common.WrapHandler(xaTransIn))
-	app.POST(BusiAPI+"/TransOutXa", common.WrapHandler(xaTransOut))
 	var err error
 	XaClient, err = dtmcli.NewXaClient(DtmServer, config.DB, Busi+"/xa", func(path string, xa *dtmcli.XaClient) {
 		app.POST(path, common.WrapHandler(func(c *gin.Context) (interface{}, error) {
@@ -38,23 +34,3 @@ func XaFireRequest() string {
 	e2p(err)
 	return gid
 }
-
-func xaTransIn(c *gin.Context) (interface{}, error) {
-	return XaClient.XaLocalTransaction(c.Request.URL.Query(), func(db *sql.DB, xa *dtmcli.Xa) (interface{}, error) {
-		if reqFrom(c).TransInResult == "FAILURE" {
-			return dtmcli.ResultFailure, nil
-		}
-		_, err := dtmcli.SdbExec(db, "update dtm_busi.user_account set balance=balance+? where user_id=?", reqFrom(c).Amount, 2)
-		return dtmcli.ResultSuccess, err
-	})
-}
-
-func xaTransOut(c *gin.Context) (interface{}, error) {
-	return XaClient.XaLocalTransaction(c.Request.URL.Query(), func(db *sql.DB, xa *dtmcli.Xa) (interface{}, error) {
-		if reqFrom(c).TransOutResult == "FAILURE" {
-			return dtmcli.ResultFailure, nil
-		}
-		_, err := dtmcli.SdbExec(db, "update dtm_busi.user_account set balance=balance-? where user_id=?", reqFrom(c).Amount, 1)
-		return dtmcli.ResultSuccess, err
-	})
-}