tsai
/
dtm
mirror of https://github.com/dtm-labs/dtm.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

rename to DoAndSubmit

pull/195/head
yedf2 5 years ago
parent
45de7d96b7
commit
556961c9f5
6 changed files with 43 additions and 33 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 27
      dtmcli/msg.go
  2. 2
      dtmgrpc/dtmgimp/types.go
  3. 26
      dtmgrpc/msg.go
  4. 1
      dtmgrpc/type.go
  5. 12
      test/msg_barrier_test.go
  6. 8
      test/msg_grpc_barrier_test.go

27
dtmcli/msg.go
View File

@ -41,30 +41,35 @@ func (s *Msg) Submit() error {
return dtmimp.TransCallDtm(&s.TransBase, s, "submit")
}
// PrepareAndSubmit one method for the entire prepare->busi->submit
func (s *Msg) PrepareAndSubmit(queryPrepared string, db *sql.DB, busiCall BarrierBusiFunc) error {
return s.Do(queryPrepared, func(bb *BranchBarrier) error {
// DoAndSubmitDB short method for Do on db type. please see Do
func (s *Msg) DoAndSubmitDB(queryPrepared string, db *sql.DB, busiCall BarrierBusiFunc) error {
return s.DoAndSubmit(queryPrepared, func(bb *BranchBarrier) error {
return bb.CallWithDB(db, busiCall)
})
}
// Do one method for the entire prepare->busi->submit
func (s *Msg) Do(queryPrepared string, busiCall func(bb *BranchBarrier) error) error {
// DoAndSubmit one method for the entire prepare->busi->submit
// if busiCall return ErrFailure, then abort is called directly
// if busiCall return not nil error other than ErrFailure, then DoAndSubmit will call queryPrepared to get the result
func (s *Msg) DoAndSubmit(queryPrepared string, busiCall func(bb *BranchBarrier) error) error {
bb, err := BarrierFrom(s.TransType, s.Gid, "00", "msg") // a special barrier for msg QueryPrepared
if err == nil {
err = s.Prepare(queryPrepared)
}
if err == nil {
err = busiCall(bb)
if err != nil && !errors.Is(err, ErrFailure) {
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)
}
if errors.Is(err, ErrFailure) {
if errors.Is(errb, ErrFailure) || errors.Is(err, ErrFailure) {
_ = dtmimp.TransCallDtm(&s.TransBase, s, "abort")
} else if err == nil {
err = s.Submit()
}
if errb != nil {
return errb
}
}
if err == nil {
err = s.Submit()
}
return err
}

2
dtmgrpc/dtmgimp/types.go
View File

@ -50,7 +50,7 @@ func GrpcClientLog(ctx context.Context, method string, req, reply interface{}, c
return err
}
// InvokeURL invoke a url for trans
// InvokeBranch invoke a url for trans
func InvokeBranch(t *dtmimp.TransBase, isRaw bool, msg proto.Message, url string, reply interface{}, branchID string, op string) error {
server, method, err := dtmdriver.GetDriver().ParseServerMethod(url)
if err != nil {

26
dtmgrpc/msg.go
View File

@ -44,31 +44,35 @@ func (s *MsgGrpc) Submit() error {
return dtmgimp.DtmGrpcCall(&s.TransBase, "Submit")
}
// PrepareAndSubmit one method for the entire prepare->busi->submit
func (s *MsgGrpc) PrepareAndSubmit(queryPrepared string, db *sql.DB, busiCall dtmcli.BarrierBusiFunc) error {
return s.Do(queryPrepared, func(bb *dtmcli.BranchBarrier) error {
// DoAndSubmitDB short method for Do on db type. please see Do
func (s *MsgGrpc) DoAndSubmitDB(queryPrepared string, db *sql.DB, busiCall dtmcli.BarrierBusiFunc) error {
return s.DoAndSubmit(queryPrepared, func(bb *dtmcli.BranchBarrier) error {
return bb.CallWithDB(db, busiCall)
})
}
// Do one method for the entire prepare->busi->submit
func (s *MsgGrpc) Do(queryPrepared string, busiCall func(bb *dtmcli.BranchBarrier) error) error {
// DoAndSubmit one method for the entire prepare->busi->submit
// if busiCall return ErrFailure, then abort is called directly
// if busiCall return not nil error other than ErrFailure, then DoAndSubmit will call queryPrepared to get the result
func (s *MsgGrpc) DoAndSubmit(queryPrepared string, busiCall func(bb *dtmcli.BranchBarrier) error) error {
bb, err := dtmcli.BarrierFrom(s.TransType, s.Gid, "00", "msg") // a special barrier for msg QueryPrepared
if err == nil {
err = s.Prepare(queryPrepared)
}
if err == nil {
err = busiCall(bb)
if err != nil && !errors.Is(err, dtmcli.ErrFailure) {
errb := busiCall(bb)
if errb != nil && !errors.Is(err, dtmcli.ErrFailure) {
err = dtmgimp.InvokeBranch(&s.TransBase, true, nil, queryPrepared, &[]byte{}, bb.BranchID, bb.Op)
err = GrpcError2DtmError(err)
}
if errors.Is(err, dtmcli.ErrFailure) {
if errors.Is(err, dtmcli.ErrFailure) || errors.Is(errb, dtmcli.ErrFailure) {
_ = dtmgimp.DtmGrpcCall(&s.TransBase, "Abort")
} else if err == nil {
err = s.Submit()
}
if errb != nil {
return errb
}
}
if err == nil {
err = s.Submit()
}
return err
}

1
dtmgrpc/type.go
View File

@ -30,6 +30,7 @@ func DtmError2GrpcError(res interface{}) error {
return e
}
// GrpcError2DtmError translate grpc error to dtm error
func GrpcError2DtmError(err error) error {
st, ok := status.FromError(err)
if ok && st.Code() == codes.Aborted {

12
test/msg_barrier_test.go
View File

@ -20,7 +20,7 @@ func TestMsgPrepareAndSubmit(t *testing.T) {
req := busi.GenTransReq(30, false, false)
msg := dtmcli.NewMsg(DtmServer, gid).
Add(busi.Busi+"/SagaBTransIn", req)
err := msg.PrepareAndSubmit(Busi+"/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
err := msg.DoAndSubmitDB(Busi+"/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
return busi.SagaAdjustBalance(tx, busi.TransOutUID, -req.Amount, "SUCCESS")
})
assert.Nil(t, err)
@ -36,7 +36,7 @@ func TestMsgPrepareAndSubmitBusiFailed(t *testing.T) {
req := busi.GenTransReq(30, false, false)
msg := dtmcli.NewMsg(DtmServer, gid).
Add(busi.Busi+"/SagaBTransIn", req)
err := msg.PrepareAndSubmit(Busi+"/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
err := msg.DoAndSubmitDB(Busi+"/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
return errors.New("an error")
})
assert.Error(t, err)
@ -49,7 +49,7 @@ func TestMsgPrepareAndSubmitPrepareFailed(t *testing.T) {
req := busi.GenTransReq(30, false, false)
msg := dtmcli.NewMsg(DtmServer+"not-exists", gid).
Add(busi.Busi+"/SagaBTransIn", req)
err := msg.PrepareAndSubmit(Busi+"/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
err := msg.DoAndSubmitDB(Busi+"/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
return busi.SagaAdjustBalance(tx, busi.TransOutUID, -req.Amount, "SUCCESS")
})
assert.Error(t, err)
@ -66,7 +66,7 @@ func TestMsgPrepareAndSubmitCommitFailed(t *testing.T) {
msg := dtmcli.NewMsg(DtmServer, gid).
Add(busi.Busi+"/SagaBTransIn", req)
var g *monkey.PatchGuard
err := msg.PrepareAndSubmit(Busi+"/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
err := msg.DoAndSubmitDB(Busi+"/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
g = monkey.PatchInstanceMethod(reflect.TypeOf(tx), "Commit", func(tx *sql.Tx) error {
logger.Debugf("tx.Commit rollback and return error in test")
_ = tx.Rollback()
@ -89,7 +89,7 @@ func TestMsgPrepareAndSubmitCommitAfterFailed(t *testing.T) {
msg := dtmcli.NewMsg(DtmServer, gid).
Add(busi.Busi+"/SagaBTransIn", req)
var guard *monkey.PatchGuard
err := msg.PrepareAndSubmit(Busi+"/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
err := msg.DoAndSubmitDB(Busi+"/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
err := busi.SagaAdjustBalance(tx, busi.TransOutUID, -req.Amount, "SUCCESS")
guard = monkey.PatchInstanceMethod(reflect.TypeOf(tx), "Commit", func(tx *sql.Tx) error {
guard.Unpatch()
@ -98,7 +98,7 @@ func TestMsgPrepareAndSubmitCommitAfterFailed(t *testing.T) {
})
return err
})
assert.Nil(t, err) // final commit will ignore error after submit
assert.Error(t, err)
waitTransProcessed(gid)
assertNotSameBalance(t, before, "mysql")
}

8
test/msg_grpc_barrier_test.go
View File

@ -20,7 +20,7 @@ func TestMsgGrpcPrepareAndSubmit(t *testing.T) {
req := busi.GenBusiReq(30, false, false)
msg := dtmgrpc.NewMsgGrpc(DtmGrpcServer, gid).
Add(busi.BusiGrpc+"/busi.Busi/TransInBSaga", req)
err := msg.PrepareAndSubmit(busi.BusiGrpc+"/busi.Busi/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
err := msg.DoAndSubmitDB(busi.BusiGrpc+"/busi.Busi/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
return busi.SagaAdjustBalance(tx, busi.TransOutUID, -int(req.Amount), "SUCCESS")
})
assert.Nil(t, err)
@ -40,7 +40,7 @@ func TestMsgGrpcPrepareAndSubmitCommitAfterFailed(t *testing.T) {
msg := dtmgrpc.NewMsgGrpc(DtmGrpcServer, gid).
Add(busi.BusiGrpc+"/busi.Busi/TransInBSaga", req)
var guard *monkey.PatchGuard
err := msg.PrepareAndSubmit(busi.BusiGrpc+"/busi.Busi/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
err := msg.DoAndSubmitDB(busi.BusiGrpc+"/busi.Busi/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
err := busi.SagaAdjustBalance(tx, busi.TransOutUID, -int(req.Amount), "SUCCESS")
guard = monkey.PatchInstanceMethod(reflect.TypeOf(tx), "Commit", func(tx *sql.Tx) error {
guard.Unpatch()
@ -49,7 +49,7 @@ func TestMsgGrpcPrepareAndSubmitCommitAfterFailed(t *testing.T) {
})
return err
})
assert.Nil(t, err)
assert.Error(t, err)
waitTransProcessed(gid)
assertNotSameBalance(t, before, "mysql")
}
@ -64,7 +64,7 @@ func TestMsgGrpcPrepareAndSubmitCommitFailed(t *testing.T) {
msg := dtmgrpc.NewMsgGrpc(DtmGrpcServer, gid).
Add(busi.Busi+"/SagaBTransIn", req)
var g *monkey.PatchGuard
err := msg.PrepareAndSubmit(busi.BusiGrpc+"/busi.Busi/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
err := msg.DoAndSubmitDB(busi.BusiGrpc+"/busi.Busi/QueryPreparedB", dbGet().ToSQLDB(), func(tx *sql.Tx) error {
g = monkey.PatchInstanceMethod(reflect.TypeOf(tx), "Commit", func(tx *sql.Tx) error {
logger.Debugf("tx.Commit rollback and return error in test")
_ = tx.Rollback()

Write Preview
Loading…
Cancel
Save