From b8127df92aaba7dfbc89f28090363e8aa7e0d736 Mon Sep 17 00:00:00 2001 From: yedf2 <120050102@qq.com> Date: Tue, 10 Aug 2021 22:29:59 +0800 Subject: [PATCH] grpc xa left a bug to fix --- dtmgrpc/xa.go | 108 +++++++++++++++++++---------------- dtmsvr/api.go | 18 ++++++ dtmsvr/api_grpc.go | 10 ++++ dtmsvr/api_http.go | 19 +----- dtmsvr/dtmsvr_test.go | 3 +- dtmsvr/trans_grpc_xa_test.go | 63 ++++++++++++++++++++ examples/busi.pb.go | 64 +++++++++++++-------- examples/busi.proto | 3 + examples/busi_grpc.pb.go | 108 +++++++++++++++++++++++++++++++++++ examples/main_base.go | 28 +++++++++ examples/main_grpc.go | 25 ++++++++ examples/main_grpc_msg.go | 6 -- examples/main_grpc_xa.go | 38 ++++++++++++ examples/main_msg.go | 5 -- examples/main_saga.go | 5 -- examples/main_saga_wait.go | 5 -- examples/main_xa.go | 24 -------- 17 files changed, 394 insertions(+), 138 deletions(-) create mode 100644 dtmsvr/trans_grpc_xa_test.go create mode 100644 examples/main_grpc_xa.go diff --git a/dtmgrpc/xa.go b/dtmgrpc/xa.go index 6884111..59caba1 100644 --- a/dtmgrpc/xa.go +++ b/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 -type XaGlobalFunc func(xa *XaGrpc) (*resty.Response, error) +// XaGrpcGlobalFunc type of xa global function +type XaGrpcGlobalFunc func(xa *XaGrpc) error -// XaLocalFunc type of xa local function -type XaLocalFunc func(db *sql.DB, xa *XaGrpc) (interface{}, error) +// XaGrpcLocalFunc type of xa local function +type XaGrpcLocalFunc func(db *sql.DB, xa *XaGrpc) error -// XaRegisterCallback type of xa register callback handler -type XaRegisterCallback func(path string, xa *XaClient) - -// XaClient xa client -type XaClient struct { +// XaGrpcClient xa client +type XaGrpcClient struct { Server string Conf map[string]string NotifyURL string @@ -31,11 +27,11 @@ type XaGrpc struct { dtmcli.TransBase } -// XaFromRequest construct xa info from request -func XaFromRequest(br *BusiRequest) (*XaGrpc, error) { +// XaGrpcFromRequest construct xa info from request +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 -func NewXaClient(server string, mysqlConf map[string]string, notifyURL string, register XaRegisterCallback) (*XaClient, error) { - xa := &XaClient{ +// NewXaGrpcClient construct a xa client +func NewXaGrpcClient(server string, mysqlConf map[string]string, notifyURL string) *XaGrpcClient { + xa := &XaGrpcClient{ Server: server, Conf: mysqlConf, NotifyURL: notifyURL, } - u, err := url.Parse(notifyURL) - if err != nil { - return nil, err - } - register(u.Path, xa) - return xa, nil + return xa } // 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) { - xa, rerr := XaFromRequest(br) +func (xc *XaGrpcClient) XaLocalTransaction(br *BusiRequest, xaFunc XaGrpcLocalFunc) (rerr error) { + 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 = dtmcli.CheckResult(ret, rerr) + rerr = xaFunc(db, xa) 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) { - xa := XaGrpc{TransBase: dtmcli.TransBase{IDGenerator: dtmcli.IDGenerator{}, Dtm: xc.Server}, TransData: dtmcli.TransData{Gid: gid, TransType: "xa"}} - rerr = xa.CallDtm(&xa.TransData, "prepare") +func (xc *XaGrpcClient) XaGlobalTransaction(gid string, xaFunc XaGrpcGlobalFunc) (rerr error) { + xa := XaGrpc{TransBase: dtmcli.TransBase{Dtm: xc.Server}, TransData: dtmcli.TransData{Gid: gid, TransType: "xa"}} + 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) - err := xa.CallDtm(&xa.TransData, operation) + if x == nil && rerr == nil { + _, 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 = dtmcli.CheckResponse(resp, rerr) + rerr = xaFunc(&xa) 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(). - SetBody(body). - SetQueryParams(dtmcli.MS{ - "gid": x.Gid, - "branch_id": branchID, - "trans_type": "xa", - "branch_type": "action", - }). - Post(url) - return resp, dtmcli.CheckResponse(resp, err) + server, method := GetServerAndMethod(url) + reply := &BusiReply{} + err := MustGetGrpcConn(server).Invoke(context.Background(), method, &BusiRequest{ + Info: &BranchInfo{ + Gid: x.Gid, + TransType: x.TransType, + BranchID: branchID, + BranchType: "", + }, + Dtm: x.Dtm, + BusiData: busiData, + }, reply) + return reply, err + } diff --git a/dtmsvr/api.go b/dtmsvr/api.go index 23d9c15..fd36ba8 100644 --- a/dtmsvr/api.go +++ b/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 +} diff --git a/dtmsvr/api_grpc.go b/dtmsvr/api_grpc.go index 7920db7..06b4e42 100644 --- a/dtmsvr/api_grpc.go +++ b/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) +} diff --git a/dtmsvr/api_http.go b/dtmsvr/api_http.go index 884d76f..27d461c 100644 --- a/dtmsvr/api_http.go +++ b/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" - 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 + return svcRegisterXaBranch(&branch) } func registerTccBranch(c *gin.Context) (interface{}, error) { diff --git a/dtmsvr/dtmsvr_test.go b/dtmsvr/dtmsvr_test.go index e050a53..67cb2e8 100644 --- a/dtmsvr/dtmsvr_test.go +++ b/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) diff --git a/dtmsvr/trans_grpc_xa_test.go b/dtmsvr/trans_grpc_xa_test.go new file mode 100644 index 0000000..7aec602 --- /dev/null +++ b/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)) +} diff --git a/examples/busi.pb.go b/examples/busi.pb.go index 58d7ce6..9cc7bfa 100644 --- a/examples/busi.pb.go +++ b/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, - 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, + 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x3a, 0x0a, 0x08, 0x58, 0x61, 0x4e, 0x6f, + 0x74, 0x69, 0x66, 0x79, 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, 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{}{ @@ -67,25 +79,31 @@ var file_examples_busi_proto_goTypes = []interface{}{ (*emptypb.Empty)(nil), // 1: google.protobuf.Empty } var file_examples_busi_proto_depIdxs = []int32{ - 0, // 0: examples.Busi.CanSubmit:input_type -> dtmgrpc.BusiRequest - 0, // 1: examples.Busi.TransIn:input_type -> dtmgrpc.BusiRequest - 0, // 2: examples.Busi.TransOut:input_type -> dtmgrpc.BusiRequest - 0, // 3: examples.Busi.TransInRevert:input_type -> dtmgrpc.BusiRequest - 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 - 1, // 8: examples.Busi.TransIn:output_type -> google.protobuf.Empty - 1, // 9: examples.Busi.TransOut:output_type -> google.protobuf.Empty - 1, // 10: examples.Busi.TransInRevert:output_type -> google.protobuf.Empty - 1, // 11: examples.Busi.TransOutRevert:output_type -> google.protobuf.Empty - 1, // 12: examples.Busi.TransInConfirm:output_type -> google.protobuf.Empty - 1, // 13: examples.Busi.TransOutConfirm:output_type -> google.protobuf.Empty - 7, // [7:14] is the sub-list for method output_type - 0, // [0:7] 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 + 0, // 0: examples.Busi.CanSubmit:input_type -> dtmgrpc.BusiRequest + 0, // 1: examples.Busi.TransIn:input_type -> dtmgrpc.BusiRequest + 0, // 2: examples.Busi.TransOut:input_type -> dtmgrpc.BusiRequest + 0, // 3: examples.Busi.TransInRevert:input_type -> dtmgrpc.BusiRequest + 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 + 0, // 7: examples.Busi.XaNotify:input_type -> dtmgrpc.BusiRequest + 0, // 8: examples.Busi.TransInXa:input_type -> dtmgrpc.BusiRequest + 0, // 9: examples.Busi.TransOutXa:input_type -> dtmgrpc.BusiRequest + 1, // 10: examples.Busi.CanSubmit:output_type -> google.protobuf.Empty + 1, // 11: examples.Busi.TransIn:output_type -> google.protobuf.Empty + 1, // 12: examples.Busi.TransOut:output_type -> google.protobuf.Empty + 1, // 13: examples.Busi.TransInRevert:output_type -> google.protobuf.Empty + 1, // 14: examples.Busi.TransOutRevert:output_type -> google.protobuf.Empty + 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 } func init() { file_examples_busi_proto_init() } diff --git a/examples/busi.proto b/examples/busi.proto index 031d8e4..deb449a 100644 --- a/examples/busi.proto +++ b/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) {} } diff --git a/examples/busi_grpc.pb.go b/examples/busi_grpc.pb.go index 39e56c6..fecfae3 100644 --- a/examples/busi_grpc.pb.go +++ b/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", diff --git a/examples/main_base.go b/examples/main_base.go index 43763a8..9e3d604 100644 --- a/examples/main_base.go +++ b/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 + }) + })) + } diff --git a/examples/main_grpc.go b/examples/main_grpc.go index 2e7a928..f0c5d01 100644 --- a/examples/main_grpc.go +++ b/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 + }) +} diff --git a/examples/main_grpc_msg.go b/examples/main_grpc_msg.go index a14ee45..4647103 100644 --- a/examples/main_grpc_msg.go +++ b/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}) diff --git a/examples/main_grpc_xa.go b/examples/main_grpc_xa.go new file mode 100644 index 0000000..27b0482 --- /dev/null +++ b/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 +} diff --git a/examples/main_msg.go b/examples/main_msg.go index c57ad54..a8527b8 100644 --- a/examples/main_msg.go +++ b/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") diff --git a/examples/main_saga.go b/examples/main_saga.go index 76f79f5..3912fc9 100644 --- a/examples/main_saga.go +++ b/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") diff --git a/examples/main_saga_wait.go b/examples/main_saga_wait.go index 53d040b..56da691 100644 --- a/examples/main_saga_wait.go +++ b/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") diff --git a/examples/main_xa.go b/examples/main_xa.go index 0d40f8f..c1783ec 100644 --- a/examples/main_xa.go +++ b/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 - }) -}