diff --git a/client/dtmcli/dtmimp/trans_base.go b/client/dtmcli/dtmimp/trans_base.go index ea2db34..16ecbd8 100644 --- a/client/dtmcli/dtmimp/trans_base.go +++ b/client/dtmcli/dtmimp/trans_base.go @@ -152,9 +152,6 @@ func TransRequestBranch(t *TransBase, method string, body interface{}, branchID SetQueryParams(query). SetHeaders(t.BranchHeaders). Execute(method, url) - if err == nil { - err = RespAsErrorCompatible(resp) - } return resp, err } diff --git a/client/dtmcli/dtmimp/utils.go b/client/dtmcli/dtmimp/utils.go index 74a0b5c..b193ea9 100644 --- a/client/dtmcli/dtmimp/utils.go +++ b/client/dtmcli/dtmimp/utils.go @@ -11,7 +11,6 @@ import ( "encoding/json" "errors" "fmt" - "net/http" "net/url" "os" "runtime" @@ -218,21 +217,6 @@ func GetDsn(conf DBConf) string { return dsn } -// RespAsErrorCompatible translate a resty response to error -// compatible with version < v1.10 -func RespAsErrorCompatible(resp *resty.Response) error { - code := resp.StatusCode() - str := resp.String() - if code == http.StatusTooEarly || strings.Contains(str, ResultOngoing) { - return fmt.Errorf("%s. %w", str, ErrOngoing) - } else if code == http.StatusConflict || strings.Contains(str, ResultFailure) { - return fmt.Errorf("%s. %w", str, ErrFailure) - } else if code != http.StatusOK { - return errors.New(str) - } - return nil -} - // RespAsErrorByJSONRPC translate json rpc resty response to error func RespAsErrorByJSONRPC(resp *resty.Response) error { str := resp.String() diff --git a/client/dtmcli/trans_msg.go b/client/dtmcli/trans_msg.go index 2304efb..ee92fbf 100644 --- a/client/dtmcli/trans_msg.go +++ b/client/dtmcli/trans_msg.go @@ -69,7 +69,7 @@ func (s *Msg) DoAndSubmit(queryPrepared string, busiCall func(bb *BranchBarrier) errb := busiCall(bb) if errb != nil && !errors.Is(errb, ErrFailure) { // if busicall return an error other than failure, we will query the result - _, err = dtmimp.TransRequestBranch(&s.TransBase, "GET", nil, bb.BranchID, bb.Op, queryPrepared) + _, err = requestBranch(&s.TransBase, "GET", nil, bb.BranchID, bb.Op, queryPrepared) } if errors.Is(errb, ErrFailure) || errors.Is(err, ErrFailure) { _ = dtmimp.TransCallDtm(&s.TransBase, "abort") diff --git a/client/dtmcli/trans_tcc.go b/client/dtmcli/trans_tcc.go index 0992e57..2c6cc72 100644 --- a/client/dtmcli/trans_tcc.go +++ b/client/dtmcli/trans_tcc.go @@ -71,5 +71,5 @@ func (t *Tcc) CallBranch(body interface{}, tryURL string, confirmURL string, can if err != nil { return nil, err } - return dtmimp.TransRequestBranch(&t.TransBase, "POST", body, branchID, dtmimp.OpTry, tryURL) + return requestBranch(&t.TransBase, "POST", body, branchID, dtmimp.OpTry, tryURL) } diff --git a/client/dtmcli/utils.go b/client/dtmcli/utils.go index 3483719..daa38be 100644 --- a/client/dtmcli/utils.go +++ b/client/dtmcli/utils.go @@ -20,22 +20,27 @@ func MustGenGid(server string) string { return res["gid"] } -// String2DtmError translate string to dtm error -func String2DtmError(str string) error { - return map[string]error{ - ResultFailure: ErrFailure, - ResultOngoing: ErrOngoing, - ResultSuccess: nil, - "": nil, - }[str] -} - -// ErrorMessage2Error return an error fmt.Errorf("%s. %w", errMsg, err) but trim out duplicate wrap +// ErrorMessage2Error return an error fmt.Errorf("%s %w", errMsg, err) but trim out duplicate wrap // eg. ErrorMessage2Error("an error. FAILURE", ErrFailure) return an error with message: "an error. FAILURE", // no additional ". FAILURE" added func ErrorMessage2Error(errMsg string, err error) error { - errMsg = strings.TrimSuffix(errMsg, ". "+err.Error()) - return fmt.Errorf("%s. %w", errMsg, err) + errMsg = strings.TrimSuffix(errMsg, " "+err.Error()) + return fmt.Errorf("%s %w", errMsg, err) +} + +// HTTPResp2DtmError translate a resty response to error +// compatible with version < v1.10 +func HTTPResp2DtmError(resp *resty.Response) error { + code := resp.StatusCode() + str := resp.String() + if code == http.StatusTooEarly || strings.Contains(str, ResultOngoing) { + return ErrorMessage2Error(str, ErrOngoing) + } else if code == http.StatusConflict || strings.Contains(str, ResultFailure) { + return ErrorMessage2Error(str, ErrFailure) + } else if code != http.StatusOK { + return errors.New(str) + } + return nil } // Result2HttpJSON return the http code and json result @@ -60,12 +65,10 @@ 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 +func requestBranch(t *dtmimp.TransBase, method string, body interface{}, branchID string, op string, url string) (*resty.Response, error) { + resp, err := dtmimp.TransRequestBranch(t, method, body, branchID, op, url) + if err == nil { + err = HTTPResp2DtmError(resp) + } + return resp, err } diff --git a/client/dtmcli/xa.go b/client/dtmcli/xa.go index cdc8c5a..251fe4b 100644 --- a/client/dtmcli/xa.go +++ b/client/dtmcli/xa.go @@ -79,5 +79,5 @@ func XaGlobalTransaction2(server string, gid string, custom func(*Xa), xaFunc Xa // CallBranch call a xa branch func (x *Xa) CallBranch(body interface{}, url string) (*resty.Response, error) { branchID := x.NewSubBranchID() - return dtmimp.TransRequestBranch(&x.TransBase, "POST", body, branchID, dtmimp.OpAction, url) + return requestBranch(&x.TransBase, "POST", body, branchID, dtmimp.OpAction, url) } diff --git a/client/workflow/imp.go b/client/workflow/imp.go index 55d08bb..7fcab39 100644 --- a/client/workflow/imp.go +++ b/client/workflow/imp.go @@ -39,7 +39,7 @@ func (wf *Workflow) loadProgresses() error { Data: p.BinData, } if sr.Status == dtmcli.StatusFailed { - sr.Error = fmt.Errorf("%s. %w", string(p.BinData), dtmcli.ErrFailure) + sr.Error = dtmcli.ErrorMessage2Error(string(p.BinData), dtmcli.ErrFailure) } wf.progresses[p.BranchID+"-"+p.Op] = sr } diff --git a/dtmsvr/api.go b/dtmsvr/api.go index 694e806..a1c80e9 100644 --- a/dtmsvr/api.go +++ b/dtmsvr/api.go @@ -123,7 +123,7 @@ func svcRegisterBranch(transType string, branch *TransBranch, data map[string]st if err == storage.ErrNotFound { msg := fmt.Sprintf("no trans with gid: %s status: %s found", branch.Gid, dtmcli.StatusPrepared) logger.Errorf(msg) - return fmt.Errorf("message: %s %w", msg, dtmcli.ErrFailure) + return dtmcli.ErrorMessage2Error(msg, dtmcli.ErrFailure) } logger.Infof("LockGlobalSaveBranches result: %v: gid: %s old status: %s branches: %s", err, branch.Gid, dtmcli.StatusPrepared, dtmimp.MustMarshalString(branches)) diff --git a/dtmsvr/trans_process.go b/dtmsvr/trans_process.go index bb4359f..7050ae3 100644 --- a/dtmsvr/trans_process.go +++ b/dtmsvr/trans_process.go @@ -49,7 +49,7 @@ func (t *TransGlobal) process(branches []TransBranch) error { if submitting && t.Status != dtmcli.StatusSucceed { if t.RollbackReason != "" { - return fmt.Errorf("%s. %w", t.RollbackReason, dtmcli.ErrFailure) + return dtmcli.ErrorMessage2Error(t.RollbackReason, dtmcli.ErrFailure) } return fmt.Errorf("wait result not return success: %w", dtmcli.ErrFailure) } diff --git a/dtmsvr/trans_status.go b/dtmsvr/trans_status.go index f0285db..6539fcc 100644 --- a/dtmsvr/trans_status.go +++ b/dtmsvr/trans_status.go @@ -135,7 +135,7 @@ func (t *TransGlobal) getHTTPResult(uri string, branchID, op string, branchPaylo if err != nil { return err } - return dtmimp.RespAsErrorCompatible(resp) + return dtmcli.HTTPResp2DtmError(resp) } func (t *TransGlobal) getJSONRPCResult(uri string, branchID, op string, branchPayload []byte) error { @@ -158,7 +158,7 @@ func (t *TransGlobal) getJSONRPCResult(uri string, branchID, op string, branchPa SetHeaders(t.TransOptions.BranchHeaders). Post(uri) if err == nil { - err = dtmimp.RespAsErrorCompatible(resp) + err = dtmcli.HTTPResp2DtmError(resp) } if err == nil { err = dtmimp.RespAsErrorByJSONRPC(resp) diff --git a/test/busi/barrier.go b/test/busi/barrier.go index 06ffd4f..ea6e53d 100644 --- a/test/busi/barrier.go +++ b/test/busi/barrier.go @@ -10,7 +10,6 @@ import ( "context" "database/sql" - "github.com/dtm-labs/dtm/client/dtmcli" "github.com/dtm-labs/dtm/client/dtmgrpc" "github.com/dtm-labs/dtm/dtmutil" "github.com/gin-gonic/gin" @@ -80,7 +79,7 @@ func init() { app.POST(BusiAPI+"/TccBTransInTry", dtmutil.WrapHandler(func(c *gin.Context) interface{} { req := reqFrom(c) if req.TransInResult != "" { - return dtmcli.String2DtmError(req.TransInResult) + return string2DtmError(req.TransInResult) } return MustBarrierFromGin(c).CallWithDB(pdbGet(), func(tx *sql.Tx) error { return tccAdjustTrading(tx, TransInUID, req.Amount) @@ -159,7 +158,7 @@ func init() { app.POST(BusiAPI+"/TccBTransOutTry", dtmutil.WrapHandler(func(c *gin.Context) interface{} { req := reqFrom(c) if req.TransOutResult != "" { - return dtmcli.String2DtmError(req.TransOutResult) + return string2DtmError(req.TransOutResult) } bb := MustBarrierFromGin(c) if req.Store == Redis { diff --git a/test/busi/base_grpc.go b/test/busi/base_grpc.go index 5d0e2f6..ee725a4 100644 --- a/test/busi/base_grpc.go +++ b/test/busi/base_grpc.go @@ -14,7 +14,6 @@ import ( "net" "time" - "github.com/dtm-labs/dtm/client/dtmcli" "github.com/dtm-labs/dtm/client/dtmcli/dtmimp" "github.com/dtm-labs/dtm/client/dtmcli/logger" "github.com/dtm-labs/dtm/client/dtmgrpc" @@ -83,7 +82,7 @@ type busiServer struct { func (s *busiServer) QueryPrepared(ctx context.Context, in *ReqGrpc) (*BusiReply, error) { res := MainSwitch.QueryPreparedResult.Fetch() - err := dtmcli.String2DtmError(res) + err := string2DtmError(res) return &BusiReply{Message: "a sample data"}, dtmgrpc.DtmError2GrpcError(err) } diff --git a/test/busi/base_http.go b/test/busi/base_http.go index 988ee6e..f5672b7 100644 --- a/test/busi/base_http.go +++ b/test/busi/base_http.go @@ -125,7 +125,7 @@ func BaseAddRoute(app *gin.Engine) { app.GET(BusiAPI+"/QueryPrepared", dtmutil.WrapHandler(func(c *gin.Context) interface{} { logger.Debugf("%s QueryPrepared", c.Query("gid")) - return dtmcli.String2DtmError(dtmimp.OrString(MainSwitch.QueryPreparedResult.Fetch(), dtmcli.ResultSuccess)) + return string2DtmError(dtmimp.OrString(MainSwitch.QueryPreparedResult.Fetch(), dtmcli.ResultSuccess)) })) app.GET(BusiAPI+"/QueryPreparedB", dtmutil.WrapHandler(func(c *gin.Context) interface{} { logger.Debugf("%s QueryPreparedB", c.Query("gid")) diff --git a/test/busi/data.go b/test/busi/data.go index a0d4a1d..4927532 100644 --- a/test/busi/data.go +++ b/test/busi/data.go @@ -67,7 +67,7 @@ func handleGeneralBusiness(c *gin.Context, result1 string, result2 string, busi if res == dtmimp.ResultFailure { return fmt.Errorf("reason:%s. %w", MainSwitch.FailureReason.Fetch(), dtmimp.ErrFailure) } - return dtmcli.String2DtmError(res) + return string2DtmError(res) } // old business handler. for compatible usage diff --git a/test/busi/utils.go b/test/busi/utils.go index 05a1055..e30b463 100644 --- a/test/busi/utils.go +++ b/test/busi/utils.go @@ -72,6 +72,16 @@ func MustBarrierFromGrpc(ctx context.Context) *dtmcli.BranchBarrier { return ti } +// string2DtmError translate string to dtm error +func string2DtmError(str string) error { + return map[string]error{ + dtmcli.ResultFailure: dtmcli.ErrFailure, + dtmcli.ResultOngoing: dtmcli.ErrOngoing, + dtmcli.ResultSuccess: nil, + "": nil, + }[str] +} + // SetGrpcHeaderForHeadersYes interceptor to set head for HeadersYes func SetGrpcHeaderForHeadersYes(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { if r, ok := req.(*dtmgpb.DtmRequest); ok && strings.HasSuffix(r.Gid, "HeadersYes") {