Browse Source

trim additional error wrap message refactored

pull/330/head
yedf2 4 years ago
parent
commit
eb6694b7c5
  1. 3
      client/dtmcli/dtmimp/trans_base.go
  2. 16
      client/dtmcli/dtmimp/utils.go
  3. 2
      client/dtmcli/trans_msg.go
  4. 2
      client/dtmcli/trans_tcc.go
  5. 45
      client/dtmcli/utils.go
  6. 2
      client/dtmcli/xa.go
  7. 2
      client/workflow/imp.go
  8. 2
      dtmsvr/api.go
  9. 2
      dtmsvr/trans_process.go
  10. 4
      dtmsvr/trans_status.go
  11. 5
      test/busi/barrier.go
  12. 3
      test/busi/base_grpc.go
  13. 2
      test/busi/base_http.go
  14. 2
      test/busi/data.go
  15. 10
      test/busi/utils.go

3
client/dtmcli/dtmimp/trans_base.go

@ -152,9 +152,6 @@ func TransRequestBranch(t *TransBase, method string, body interface{}, branchID
SetQueryParams(query). SetQueryParams(query).
SetHeaders(t.BranchHeaders). SetHeaders(t.BranchHeaders).
Execute(method, url) Execute(method, url)
if err == nil {
err = RespAsErrorCompatible(resp)
}
return resp, err return resp, err
} }

16
client/dtmcli/dtmimp/utils.go

@ -11,7 +11,6 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"net/http"
"net/url" "net/url"
"os" "os"
"runtime" "runtime"
@ -218,21 +217,6 @@ func GetDsn(conf DBConf) string {
return dsn 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 // RespAsErrorByJSONRPC translate json rpc resty response to error
func RespAsErrorByJSONRPC(resp *resty.Response) error { func RespAsErrorByJSONRPC(resp *resty.Response) error {
str := resp.String() str := resp.String()

2
client/dtmcli/trans_msg.go

@ -69,7 +69,7 @@ func (s *Msg) DoAndSubmit(queryPrepared string, busiCall func(bb *BranchBarrier)
errb := busiCall(bb) errb := busiCall(bb)
if errb != nil && !errors.Is(errb, ErrFailure) { if errb != nil && !errors.Is(errb, ErrFailure) {
// if busicall return an error other than failure, we will query the result // 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) { if errors.Is(errb, ErrFailure) || errors.Is(err, ErrFailure) {
_ = dtmimp.TransCallDtm(&s.TransBase, "abort") _ = dtmimp.TransCallDtm(&s.TransBase, "abort")

2
client/dtmcli/trans_tcc.go

@ -71,5 +71,5 @@ func (t *Tcc) CallBranch(body interface{}, tryURL string, confirmURL string, can
if err != nil { if err != nil {
return nil, err return nil, err
} }
return dtmimp.TransRequestBranch(&t.TransBase, "POST", body, branchID, dtmimp.OpTry, tryURL) return requestBranch(&t.TransBase, "POST", body, branchID, dtmimp.OpTry, tryURL)
} }

45
client/dtmcli/utils.go

@ -20,22 +20,27 @@ func MustGenGid(server string) string {
return res["gid"] return res["gid"]
} }
// String2DtmError translate string to dtm error // ErrorMessage2Error return an error fmt.Errorf("%s %w", errMsg, err) but trim out duplicate wrap
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
// eg. ErrorMessage2Error("an error. FAILURE", ErrFailure) return an error with message: "an error. FAILURE", // eg. ErrorMessage2Error("an error. FAILURE", ErrFailure) return an error with message: "an error. FAILURE",
// no additional ". FAILURE" added // no additional ". FAILURE" added
func ErrorMessage2Error(errMsg string, err error) error { func ErrorMessage2Error(errMsg string, err error) error {
errMsg = strings.TrimSuffix(errMsg, ". "+err.Error()) errMsg = strings.TrimSuffix(errMsg, " "+err.Error())
return fmt.Errorf("%s. %w", errMsg, err) 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 // Result2HttpJSON return the http code and json result
@ -60,12 +65,10 @@ func Result2HttpJSON(result interface{}) (code int, res interface{}) {
return return
} }
// IsRollback returns whether the result is indicating rollback func requestBranch(t *dtmimp.TransBase, method string, body interface{}, branchID string, op string, url string) (*resty.Response, error) {
func IsRollback(resp *resty.Response, err error) bool { resp, err := dtmimp.TransRequestBranch(t, method, body, branchID, op, url)
return err == ErrFailure || dtmimp.RespAsErrorCompatible(resp) == ErrFailure if err == nil {
} err = HTTPResp2DtmError(resp)
}
// IsOngoing returns whether the result is indicating ongoing return resp, err
func IsOngoing(resp *resty.Response, err error) bool {
return err == ErrOngoing || dtmimp.RespAsErrorCompatible(resp) == ErrOngoing
} }

2
client/dtmcli/xa.go

@ -79,5 +79,5 @@ func XaGlobalTransaction2(server string, gid string, custom func(*Xa), xaFunc Xa
// CallBranch call a xa branch // CallBranch call a xa branch
func (x *Xa) CallBranch(body interface{}, url string) (*resty.Response, error) { func (x *Xa) CallBranch(body interface{}, url string) (*resty.Response, error) {
branchID := x.NewSubBranchID() branchID := x.NewSubBranchID()
return dtmimp.TransRequestBranch(&x.TransBase, "POST", body, branchID, dtmimp.OpAction, url) return requestBranch(&x.TransBase, "POST", body, branchID, dtmimp.OpAction, url)
} }

2
client/workflow/imp.go

@ -39,7 +39,7 @@ func (wf *Workflow) loadProgresses() error {
Data: p.BinData, Data: p.BinData,
} }
if sr.Status == dtmcli.StatusFailed { 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 wf.progresses[p.BranchID+"-"+p.Op] = sr
} }

2
dtmsvr/api.go

@ -123,7 +123,7 @@ func svcRegisterBranch(transType string, branch *TransBranch, data map[string]st
if err == storage.ErrNotFound { if err == storage.ErrNotFound {
msg := fmt.Sprintf("no trans with gid: %s status: %s found", branch.Gid, dtmcli.StatusPrepared) msg := fmt.Sprintf("no trans with gid: %s status: %s found", branch.Gid, dtmcli.StatusPrepared)
logger.Errorf(msg) 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", logger.Infof("LockGlobalSaveBranches result: %v: gid: %s old status: %s branches: %s",
err, branch.Gid, dtmcli.StatusPrepared, dtmimp.MustMarshalString(branches)) err, branch.Gid, dtmcli.StatusPrepared, dtmimp.MustMarshalString(branches))

2
dtmsvr/trans_process.go

@ -49,7 +49,7 @@ func (t *TransGlobal) process(branches []TransBranch) error {
if submitting && t.Status != dtmcli.StatusSucceed { if submitting && t.Status != dtmcli.StatusSucceed {
if t.RollbackReason != "" { 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) return fmt.Errorf("wait result not return success: %w", dtmcli.ErrFailure)
} }

4
dtmsvr/trans_status.go

@ -135,7 +135,7 @@ func (t *TransGlobal) getHTTPResult(uri string, branchID, op string, branchPaylo
if err != nil { if err != nil {
return err return err
} }
return dtmimp.RespAsErrorCompatible(resp) return dtmcli.HTTPResp2DtmError(resp)
} }
func (t *TransGlobal) getJSONRPCResult(uri string, branchID, op string, branchPayload []byte) error { 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). SetHeaders(t.TransOptions.BranchHeaders).
Post(uri) Post(uri)
if err == nil { if err == nil {
err = dtmimp.RespAsErrorCompatible(resp) err = dtmcli.HTTPResp2DtmError(resp)
} }
if err == nil { if err == nil {
err = dtmimp.RespAsErrorByJSONRPC(resp) err = dtmimp.RespAsErrorByJSONRPC(resp)

5
test/busi/barrier.go

@ -10,7 +10,6 @@ import (
"context" "context"
"database/sql" "database/sql"
"github.com/dtm-labs/dtm/client/dtmcli"
"github.com/dtm-labs/dtm/client/dtmgrpc" "github.com/dtm-labs/dtm/client/dtmgrpc"
"github.com/dtm-labs/dtm/dtmutil" "github.com/dtm-labs/dtm/dtmutil"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
@ -80,7 +79,7 @@ func init() {
app.POST(BusiAPI+"/TccBTransInTry", dtmutil.WrapHandler(func(c *gin.Context) interface{} { app.POST(BusiAPI+"/TccBTransInTry", dtmutil.WrapHandler(func(c *gin.Context) interface{} {
req := reqFrom(c) req := reqFrom(c)
if req.TransInResult != "" { if req.TransInResult != "" {
return dtmcli.String2DtmError(req.TransInResult) return string2DtmError(req.TransInResult)
} }
return MustBarrierFromGin(c).CallWithDB(pdbGet(), func(tx *sql.Tx) error { return MustBarrierFromGin(c).CallWithDB(pdbGet(), func(tx *sql.Tx) error {
return tccAdjustTrading(tx, TransInUID, req.Amount) return tccAdjustTrading(tx, TransInUID, req.Amount)
@ -159,7 +158,7 @@ func init() {
app.POST(BusiAPI+"/TccBTransOutTry", dtmutil.WrapHandler(func(c *gin.Context) interface{} { app.POST(BusiAPI+"/TccBTransOutTry", dtmutil.WrapHandler(func(c *gin.Context) interface{} {
req := reqFrom(c) req := reqFrom(c)
if req.TransOutResult != "" { if req.TransOutResult != "" {
return dtmcli.String2DtmError(req.TransOutResult) return string2DtmError(req.TransOutResult)
} }
bb := MustBarrierFromGin(c) bb := MustBarrierFromGin(c)
if req.Store == Redis { if req.Store == Redis {

3
test/busi/base_grpc.go

@ -14,7 +14,6 @@ import (
"net" "net"
"time" "time"
"github.com/dtm-labs/dtm/client/dtmcli"
"github.com/dtm-labs/dtm/client/dtmcli/dtmimp" "github.com/dtm-labs/dtm/client/dtmcli/dtmimp"
"github.com/dtm-labs/dtm/client/dtmcli/logger" "github.com/dtm-labs/dtm/client/dtmcli/logger"
"github.com/dtm-labs/dtm/client/dtmgrpc" "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) { func (s *busiServer) QueryPrepared(ctx context.Context, in *ReqGrpc) (*BusiReply, error) {
res := MainSwitch.QueryPreparedResult.Fetch() res := MainSwitch.QueryPreparedResult.Fetch()
err := dtmcli.String2DtmError(res) err := string2DtmError(res)
return &BusiReply{Message: "a sample data"}, dtmgrpc.DtmError2GrpcError(err) return &BusiReply{Message: "a sample data"}, dtmgrpc.DtmError2GrpcError(err)
} }

2
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{} { app.GET(BusiAPI+"/QueryPrepared", dtmutil.WrapHandler(func(c *gin.Context) interface{} {
logger.Debugf("%s QueryPrepared", c.Query("gid")) 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{} { app.GET(BusiAPI+"/QueryPreparedB", dtmutil.WrapHandler(func(c *gin.Context) interface{} {
logger.Debugf("%s QueryPreparedB", c.Query("gid")) logger.Debugf("%s QueryPreparedB", c.Query("gid"))

2
test/busi/data.go

@ -67,7 +67,7 @@ func handleGeneralBusiness(c *gin.Context, result1 string, result2 string, busi
if res == dtmimp.ResultFailure { if res == dtmimp.ResultFailure {
return fmt.Errorf("reason:%s. %w", MainSwitch.FailureReason.Fetch(), dtmimp.ErrFailure) return fmt.Errorf("reason:%s. %w", MainSwitch.FailureReason.Fetch(), dtmimp.ErrFailure)
} }
return dtmcli.String2DtmError(res) return string2DtmError(res)
} }
// old business handler. for compatible usage // old business handler. for compatible usage

10
test/busi/utils.go

@ -72,6 +72,16 @@ func MustBarrierFromGrpc(ctx context.Context) *dtmcli.BranchBarrier {
return ti 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 // 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 { 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") { if r, ok := req.(*dtmgpb.DtmRequest); ok && strings.HasSuffix(r.Gid, "HeadersYes") {

Loading…
Cancel
Save