From f1a2a38a60701c020a382f9d1c01f3d6ea23c378 Mon Sep 17 00:00:00 2001 From: yedf2 <120050102@qq.com> Date: Wed, 29 Jun 2022 21:44:16 +0800 Subject: [PATCH] fix lint --- dtmcli/dtmimp/trans_base.go | 1 + dtmcli/dtmimp/vars.go | 2 ++ dtmcli/utils.go | 2 ++ dtmgrpc/workflow/imp.go | 12 ++++++------ dtmgrpc/workflow/utils.go | 13 +++++-------- dtmgrpc/workflow/workflow.go | 18 ++++++++++-------- dtmsvr/storage/boltdb/boltdb.go | 9 +++++++-- test/busi/base_grpc.go | 1 + test/busi/base_http.go | 2 +- test/busi/base_types.go | 16 ++++++++-------- test/busi/base_workflow.go | 3 ++- test/busi/utils.go | 1 + test/tcc_barrier_test.go | 2 +- test/workflow_test.go | 8 ++++---- 14 files changed, 51 insertions(+), 39 deletions(-) diff --git a/dtmcli/dtmimp/trans_base.go b/dtmcli/dtmimp/trans_base.go index 5e696c7..ea2db34 100644 --- a/dtmcli/dtmimp/trans_base.go +++ b/dtmcli/dtmimp/trans_base.go @@ -113,6 +113,7 @@ func TransCallDtmExt(tb *TransBase, body interface{}, operation string) (*resty. return resp, nil } +// TransCallDtm is the short call for TransCallDtmExt func TransCallDtm(tb *TransBase, operation string) error { _, err := TransCallDtmExt(tb, tb, operation) return err diff --git a/dtmcli/dtmimp/vars.go b/dtmcli/dtmimp/vars.go index 7bb036a..4110177 100644 --- a/dtmcli/dtmimp/vars.go +++ b/dtmcli/dtmimp/vars.go @@ -42,6 +42,7 @@ var PassthroughHeaders = []string{} // BarrierTableName the table name of barrier table var BarrierTableName = "dtm_barrier.barrier" +// BeforeRequest is the middleware for default resty.Client func BeforeRequest(c *resty.Client, r *resty.Request) error { r.URL = MayReplaceLocalhost(r.URL) u, err := dtmdriver.GetHTTPDriver().ResolveURL(r.URL) @@ -50,6 +51,7 @@ func BeforeRequest(c *resty.Client, r *resty.Request) error { return err } +// AfterResponse is the middleware for default resty.Client func AfterResponse(c *resty.Client, resp *resty.Response) error { r := resp.Request logger.Debugf("requested: %d %s %s %s", resp.StatusCode(), r.Method, r.URL, resp.String()) diff --git a/dtmcli/utils.go b/dtmcli/utils.go index 63ad69b..51dde87 100644 --- a/dtmcli/utils.go +++ b/dtmcli/utils.go @@ -51,10 +51,12 @@ func Result2HttpJSON(result interface{}) (code int, res interface{}) { return } +// IsRollback returns whether the result is indicating rollback func IsRollback(resp *resty.Response, err error) bool { return err == ErrFailure || dtmimp.RespAsErrorCompatible(resp) == ErrFailure } +// IsOngoing returns whether the result is indicating ongoing func IsOngoing(resp *resty.Response, err error) bool { return err == ErrOngoing || dtmimp.RespAsErrorCompatible(resp) == ErrOngoing } diff --git a/dtmgrpc/workflow/imp.go b/dtmgrpc/workflow/imp.go index bea6fad..7f43199 100644 --- a/dtmgrpc/workflow/imp.go +++ b/dtmgrpc/workflow/imp.go @@ -12,10 +12,10 @@ import ( ) type workflowImp struct { - restyClient *resty.Client + restyClient *resty.Client //nolint idGen dtmimp.BranchIDGen - currentBranch string - progresses map[string]*stepResult + currentBranch string //nolint + progresses map[string]*stepResult //nolint currentOp string succeededOps []workflowPhase2Item failedOps []workflowPhase2Item @@ -84,7 +84,7 @@ func (wf *Workflow) initRestyClient() { return err }) old := wf.restyClient.GetClient().Transport - wf.restyClient.GetClient().Transport = NewRoundTripper(old, wf) + wf.restyClient.GetClient().Transport = newRoundTripper(old, wf) wf.restyClient.OnAfterResponse(func(c *resty.Client, r *resty.Response) error { err := dtmimp.AfterResponse(c, r) if err == nil && !wf.Options.DisalbeAutoError { @@ -135,7 +135,7 @@ func (wf *Workflow) processPhase2(err error) error { for i := len(ops) - 1; i >= 0; i-- { op := ops[i] - err1 := wf.callPhase2(op.branchID, op.op, op.fn) + err1 := wf.callPhase2(op.branchID, op.fn) if err1 != nil { return err1 } @@ -143,7 +143,7 @@ func (wf *Workflow) processPhase2(err error) error { return err } -func (wf *Workflow) callPhase2(branchID string, op string, fn WfPhase2Func) error { +func (wf *Workflow) callPhase2(branchID string, fn WfPhase2Func) error { wf.currentBranch = branchID r := wf.recordedDo(func(bb *dtmcli.BranchBarrier) *stepResult { err := fn(bb) diff --git a/dtmgrpc/workflow/utils.go b/dtmgrpc/workflow/utils.go index ae0a0d3..fcadcd6 100644 --- a/dtmgrpc/workflow/utils.go +++ b/dtmgrpc/workflow/utils.go @@ -15,9 +15,6 @@ import ( "google.golang.org/protobuf/reflect/protoreflect" ) -const HBranchID = "dtm-branch-id" -const HBranchOp = "dtm-branch-op" - func statusToCode(status string) int { if status == "succeed" { return 200 @@ -66,12 +63,12 @@ func (r *roundTripper) RoundTrip(req *http.Request) (*http.Response, error) { } sr := wf.recordedDo(func(bb *dtmcli.BranchBarrier) *stepResult { resp, err := r.old.RoundTrip(req) - return stepResultFromHttp(resp, err) + return stepResultFromHTTP(resp, err) }) - return stepResultToHttp(sr) + return stepResultToHTTP(sr) } -func NewRoundTripper(old http.RoundTripper, wf *Workflow) http.RoundTripper { +func newRoundTripper(old http.RoundTripper, wf *Workflow) http.RoundTripper { return &roundTripper{old: old, wf: wf} } @@ -117,7 +114,7 @@ func stepResultToGrpc(s *stepResult, reply interface{}) error { return status.New(codes.Aborted, string(s.Data)).Err() } -func stepResultFromHttp(resp *http.Response, err error) *stepResult { +func stepResultFromHTTP(resp *http.Response, err error) *stepResult { sr := &stepResult{Error: err} if err == nil { sr.Data, sr.Error = ioutil.ReadAll(resp.Body) @@ -130,7 +127,7 @@ func stepResultFromHttp(resp *http.Response, err error) *stepResult { return sr } -func stepResultToHttp(s *stepResult) (*http.Response, error) { +func stepResultToHTTP(s *stepResult) (*http.Response, error) { if s.Error != nil { return nil, s.Error } diff --git a/dtmgrpc/workflow/workflow.go b/dtmgrpc/workflow/workflow.go index ed6f714..5c1fc20 100644 --- a/dtmgrpc/workflow/workflow.go +++ b/dtmgrpc/workflow/workflow.go @@ -15,16 +15,16 @@ import ( "google.golang.org/grpc" ) -// InitHttp will init Workflow engine to use http +// InitHTTP will init Workflow engine to use http // param httpDtm specify the dtm address // param callback specify the url for dtm to callback if a workflow timeout -func InitHttp(httpDtm string, callback string) { +func InitHTTP(httpDtm string, callback string) { defaultFac.protocol = dtmimp.ProtocolHTTP defaultFac.httpDtm = httpDtm defaultFac.httpCallback = callback } -// InitHttp will init Workflow engine to use grpc +// InitGrpc will init Workflow engine to use grpc // param dtm specify the dtm address // param clientHost specify the client host for dtm to callback if a workflow timeout // param grpcServer specify the grpc server @@ -57,10 +57,10 @@ func ExecuteByQS(qs url.Values, body []byte) error { return defaultFac.executeByQS(qs, body) } -// WorkflowOptions is for specifying workflow options -type WorkflowOptions struct { +// Options is for specifying workflow options +type Options struct { // if this flag is set true, then Workflow's restyClient will keep the origin http response - // or else, Workflow's restyClient will convert http reponse to error if status code is not 200 + // or else, Workflow's restyClient will convert http response to error if status code is not 200 DisalbeAutoError bool } @@ -68,7 +68,7 @@ type WorkflowOptions struct { type Workflow struct { // The name of the workflow Name string - Options WorkflowOptions + Options Options *dtmimp.TransBase workflowImp } @@ -96,7 +96,7 @@ func (wf *Workflow) AddSagaPhase2(compensate WfPhase2Func) { }) } -// DefineSagaPhase2 will define a tcc branch transaction +// AddTccPhase2 will define a tcc branch transaction // param confirm, concel specify the confirm and cancel operation of next workflow action func (wf *Workflow) AddTccPhase2(confirm, cancel WfPhase2Func) { branchID := wf.currentBranch @@ -121,6 +121,8 @@ func (wf *Workflow) DoAction(fn func(bb *dtmcli.BranchBarrier) ([]byte, error)) return stepResultToLocal(res) } +// DoXaAction will begin a local xa transaction +// after the return of workflow function, xa commit/rollback will be called func (wf *Workflow) DoXaAction(dbConf dtmcli.DBConf, fn func(db *sql.DB) ([]byte, error)) ([]byte, error) { branchID := wf.currentBranch res := wf.recordedDo(func(bb *dtmcli.BranchBarrier) *stepResult { diff --git a/dtmsvr/storage/boltdb/boltdb.go b/dtmsvr/storage/boltdb/boltdb.go index c1f646d..65571af 100644 --- a/dtmsvr/storage/boltdb/boltdb.go +++ b/dtmsvr/storage/boltdb/boltdb.go @@ -209,7 +209,12 @@ func tPutGlobal(t *bolt.Tx, global *storage.TransGlobalStore) { dtmimp.E2P(err) } -func tPutBranches(t *bolt.Tx, branches []storage.TransBranchStore, start int64) error { +func tPutBranches(t *bolt.Tx, branches []storage.TransBranchStore, start int64) { + err := tPutBranches2(t, branches, start) + dtmimp.E2P(err) +} + +func tPutBranches2(t *bolt.Tx, branches []storage.TransBranchStore, start int64) error { if start == -1 { b0 := &branches[0] bs := tGetBranches(t, b0.Gid) @@ -330,7 +335,7 @@ func (s *Store) LockGlobalSaveBranches(gid string, status string, branches []sto if g.Status != status { return storage.ErrNotFound } - return tPutBranches(t, branches, int64(branchStart)) + return tPutBranches2(t, branches, int64(branchStart)) }) dtmimp.E2P(err) } diff --git a/test/busi/base_grpc.go b/test/busi/base_grpc.go index d52d907..a00d101 100644 --- a/test/busi/base_grpc.go +++ b/test/busi/base_grpc.go @@ -33,6 +33,7 @@ var BusiGrpc = fmt.Sprintf("localhost:%d", BusiGrpcPort) // DtmClient grpc client for dtm var DtmClient dtmgpb.DtmClient +// BusiCli grpc client for busi var BusiCli BusiClient // GrpcStartup for grpc diff --git a/test/busi/base_http.go b/test/busi/base_http.go index d0d6a4b..056947a 100644 --- a/test/busi/base_http.go +++ b/test/busi/base_http.go @@ -158,7 +158,7 @@ func BaseAddRoute(app *gin.Engine) { tcc, err := dtmcli.TccFromQuery(c.Request.URL.Query()) logger.FatalIfError(err) logger.Debugf("TransInTccNested ") - resp, err := tcc.CallBranch(&ReqHttp{Amount: reqFrom(c).Amount}, Busi+"/TransIn", Busi+"/TransInConfirm", Busi+"/TransInRevert") + resp, err := tcc.CallBranch(&ReqHTTP{Amount: reqFrom(c).Amount}, Busi+"/TransIn", Busi+"/TransInConfirm", Busi+"/TransInRevert") if err != nil { return err } diff --git a/test/busi/base_types.go b/test/busi/base_types.go index 9cb1c28..42dcc9e 100644 --- a/test/busi/base_types.go +++ b/test/busi/base_types.go @@ -60,21 +60,21 @@ func GetBalanceByUID(uid int, store string) int { return dtmimp.MustAtoi(ua.Balance[:len(ua.Balance)-3]) } -// ReqHttp transaction request payload -type ReqHttp struct { +// ReqHTTP transaction request payload +type ReqHTTP struct { Amount int `json:"amount"` TransInResult string `json:"trans_in_result"` TransOutResult string `json:"trans_out_Result"` Store string `json:"store"` // default mysql, value can be mysql|redis } -func (t *ReqHttp) String() string { +func (t *ReqHTTP) String() string { return fmt.Sprintf("amount: %d transIn: %s transOut: %s", t.Amount, t.TransInResult, t.TransOutResult) } // GenTransReq 1 -func GenTransReq(amount int, outFailed bool, inFailed bool) *ReqHttp { - return &ReqHttp{ +func GenTransReq(amount int, outFailed bool, inFailed bool) *ReqHTTP { + return &ReqHTTP{ Amount: amount, TransOutResult: dtmimp.If(outFailed, dtmcli.ResultFailure, "").(string), TransInResult: dtmimp.If(inFailed, dtmcli.ResultFailure, "").(string), @@ -90,16 +90,16 @@ func GenBusiReq(amount int, outFailed bool, inFailed bool) *BusiReq { } } -func reqFrom(c *gin.Context) *ReqHttp { +func reqFrom(c *gin.Context) *ReqHTTP { v, ok := c.Get("trans_req") if !ok { - req := ReqHttp{} + req := ReqHTTP{} err := c.BindJSON(&req) logger.FatalIfError(err) c.Set("trans_req", &req) v = &req } - return v.(*ReqHttp) + return v.(*ReqHTTP) } func infoFromContext(c *gin.Context) *dtmcli.BranchBarrier { diff --git a/test/busi/base_workflow.go b/test/busi/base_workflow.go index bb44a8a..e485e0c 100644 --- a/test/busi/base_workflow.go +++ b/test/busi/base_workflow.go @@ -6,7 +6,8 @@ import ( "google.golang.org/grpc" ) +// WorkflowStarup 1 func WorkflowStarup(server *grpc.Server) { - workflow.InitHttp(dtmServer, Busi+"/workflow/resume") + workflow.InitHTTP(dtmServer, Busi+"/workflow/resume") workflow.InitGrpc(dtmutil.DefaultGrpcServer, BusiGrpc, server) } diff --git a/test/busi/utils.go b/test/busi/utils.go index 49b02e2..d8768d3 100644 --- a/test/busi/utils.go +++ b/test/busi/utils.go @@ -25,6 +25,7 @@ import ( "google.golang.org/grpc/metadata" ) +// ReqGrpc is the req for grpc protocol type ReqGrpc = BusiReq func dbGet() *dtmutil.DB { diff --git a/test/tcc_barrier_test.go b/test/tcc_barrier_test.go index bfa5e90..9246e44 100644 --- a/test/tcc_barrier_test.go +++ b/test/tcc_barrier_test.go @@ -69,7 +69,7 @@ func runTestTccBarrierDisorder(t *testing.T, store string) { gid := dtmimp.GetFuncName() + store cronFinished := make(chan string, 2) err := dtmcli.TccGlobalTransaction(DtmServer, gid, func(tcc *dtmcli.Tcc) (*resty.Response, error) { - body := &busi.ReqHttp{Amount: 30, Store: store} + body := &busi.ReqHTTP{Amount: 30, Store: store} tryURL := Busi + "/TccBTransOutTry" confirmURL := Busi + "/TccBTransOutConfirm" cancelURL := Busi + "/SleepCancel" diff --git a/test/workflow_test.go b/test/workflow_test.go index 0fcfa92..b701b32 100644 --- a/test/workflow_test.go +++ b/test/workflow_test.go @@ -28,7 +28,7 @@ func TestWorkflowNormal(t *testing.T) { gid := dtmimp.GetFuncName() workflow.Register(gid, func(wf *workflow.Workflow, data []byte) error { - var req busi.ReqHttp + var req busi.ReqHTTP dtmimp.MustUnmarshal(data, &req) _, err := wf.NewRequest().SetBody(req).Post(Busi + "/TransOut") if err != nil { @@ -50,11 +50,11 @@ func TestWorkflowNormal(t *testing.T) { func TestWorkflowRollback(t *testing.T) { workflow.SetProtocolForTest(dtmimp.ProtocolHTTP) - req := &busi.ReqHttp{Amount: 30, TransInResult: dtmimp.ResultFailure} + req := &busi.ReqHTTP{Amount: 30, TransInResult: dtmimp.ResultFailure} gid := dtmimp.GetFuncName() workflow.Register(gid, func(wf *workflow.Workflow, data []byte) error { - var req busi.ReqHttp + var req busi.ReqHTTP dtmimp.MustUnmarshal(data, &req) wf.AddSagaPhase2(func(bb *dtmcli.BranchBarrier) error { _, err := wf.NewRequest().SetBody(req).Post(Busi + "/SagaBTransOutCom") @@ -313,7 +313,7 @@ func TestWorkflowMixed(t *testing.T) { _, err := busi.BusiCli.TransInConfirm(wf.Context, &req) return err }, func(bb *dtmcli.BranchBarrier) error { - req2 := &busi.ReqHttp{Amount: 30} + req2 := &busi.ReqHTTP{Amount: 30} _, err := wf.NewRequest().SetBody(req2).Post(Busi + "/TransInRevert") return err })