mirror of https://github.com/dtm-labs/dtm.git
yedf2
5 years ago
committed by
GitHub
GitHub
21 changed files with 440 additions and 71 deletions
@ -0,0 +1,102 @@ |
|||
package dtmcli |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
"strings" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/dtmcli/logger" |
|||
"go.mongodb.org/mongo-driver/bson" |
|||
"go.mongodb.org/mongo-driver/mongo" |
|||
) |
|||
|
|||
// MongoCall sub-trans barrier for mongo. see http://dtm.pub/practice/barrier
|
|||
// experimental
|
|||
func (bb *BranchBarrier) MongoCall(mc *mongo.Client, busiCall func(mongo.SessionContext) error) (rerr error) { |
|||
bb.BarrierID = bb.BarrierID + 1 |
|||
bid := fmt.Sprintf("%02d", bb.BarrierID) |
|||
return mc.UseSession(context.Background(), func(sc mongo.SessionContext) (rerr error) { |
|||
rerr = sc.StartTransaction() |
|||
if rerr != nil { |
|||
return nil |
|||
} |
|||
defer dtmimp.DeferDo(&rerr, func() error { |
|||
return sc.CommitTransaction(sc) |
|||
}, func() error { |
|||
return sc.AbortTransaction(sc) |
|||
}) |
|||
originOp := map[string]string{ |
|||
BranchCancel: BranchTry, |
|||
BranchCompensate: BranchAction, |
|||
}[bb.Op] |
|||
|
|||
originAffected, oerr := mongoInsertBarrier(sc, mc, bb.TransType, bb.Gid, bb.BranchID, originOp, bid, bb.Op) |
|||
currentAffected, rerr := mongoInsertBarrier(sc, mc, bb.TransType, bb.Gid, bb.BranchID, bb.Op, bid, bb.Op) |
|||
logger.Debugf("originAffected: %d currentAffected: %d", originAffected, currentAffected) |
|||
if rerr == nil { |
|||
rerr = oerr |
|||
} |
|||
if (bb.Op == BranchCancel || bb.Op == BranchCompensate) && originAffected > 0 || // null compensate
|
|||
currentAffected == 0 { // repeated request or dangled request
|
|||
return |
|||
} |
|||
if rerr == nil { |
|||
rerr = busiCall(sc) |
|||
} |
|||
return |
|||
}) |
|||
} |
|||
|
|||
// MongoQueryPrepared query prepared for redis
|
|||
// experimental
|
|||
func (bb *BranchBarrier) MongoQueryPrepared(mc *mongo.Client) error { |
|||
_, err := mongoInsertBarrier(context.Background(), mc, bb.TransType, bb.Gid, "00", "msg", "01", "rollback") |
|||
var result bson.M |
|||
if err == nil { |
|||
fs := strings.Split(dtmimp.BarrierTableName, ".") |
|||
barrier := mc.Database(fs[0]).Collection(fs[1]) |
|||
err = barrier.FindOne(context.Background(), bson.D{ |
|||
{Key: "gid", Value: bb.Gid}, |
|||
{Key: "branch_id", Value: "00"}, |
|||
{Key: "op", Value: "msg"}, |
|||
{Key: "barrier_id", Value: "01"}, |
|||
}).Decode(&result) |
|||
} |
|||
var reason string |
|||
if err == nil { |
|||
reason, _ = result["reason"].(string) |
|||
} |
|||
if err == nil && reason == "rollback" { |
|||
return ErrFailure |
|||
} |
|||
return err |
|||
} |
|||
|
|||
func mongoInsertBarrier(sc context.Context, mc *mongo.Client, transType string, gid string, branchID string, op string, barrierID string, reason string) (int64, error) { |
|||
if op == "" { |
|||
return 0, nil |
|||
} |
|||
fs := strings.Split(dtmimp.BarrierTableName, ".") |
|||
barrier := mc.Database(fs[0]).Collection(fs[1]) |
|||
r := barrier.FindOne(sc, bson.D{ |
|||
{Key: "gid", Value: gid}, |
|||
{Key: "branch_id", Value: branchID}, |
|||
{Key: "op", Value: op}, |
|||
{Key: "barrier_id", Value: barrierID}, |
|||
}) |
|||
err := r.Err() |
|||
if err == mongo.ErrNoDocuments { |
|||
_, err = barrier.InsertOne(sc, |
|||
bson.D{ |
|||
{Key: "trans_type", Value: transType}, |
|||
{Key: "gid", Value: gid}, |
|||
{Key: "branch_id", Value: branchID}, |
|||
{Key: "op", Value: op}, |
|||
{Key: "barrier_id", Value: barrierID}, |
|||
{Key: "reason", Value: reason}, |
|||
}) |
|||
return 1, err |
|||
} |
|||
return 0, err |
|||
} |
|||
@ -0,0 +1,3 @@ |
|||
use busi |
|||
db.busi.insert({user_id: 1, balance: 10000}) |
|||
db.busi.insert({user_id: 2, balance: 10000}) |
|||
@ -0,0 +1,4 @@ |
|||
use dtm_barrier |
|||
db.barrier.drop() |
|||
db.barrier.createIndex({gid:1, branch_id:1, op: 1, barrier_id: 1}, {unique: true}) |
|||
//db.barrier.insert({gid:"123", branch_id:"01", op:"action", barrier_id:"01", reason:"action"});
|
|||
@ -0,0 +1,78 @@ |
|||
package test |
|||
|
|||
import ( |
|||
"errors" |
|||
"testing" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli" |
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/test/busi" |
|||
"github.com/stretchr/testify/assert" |
|||
"go.mongodb.org/mongo-driver/mongo" |
|||
) |
|||
|
|||
func TestMsgMongoDoSucceed(t *testing.T) { |
|||
before := getBeforeBalances("mongo") |
|||
gid := dtmimp.GetFuncName() |
|||
req := busi.GenTransReq(30, false, false) |
|||
msg := dtmcli.NewMsg(DtmServer, gid). |
|||
Add(busi.Busi+"/SagaMongoTransIn", req) |
|||
err := msg.DoAndSubmit(Busi+"/MongoQueryPrepared", func(bb *dtmcli.BranchBarrier) error { |
|||
return bb.MongoCall(busi.MongoGet(), func(sc mongo.SessionContext) error { |
|||
return busi.SagaMongoAdjustBalance(sc, sc.Client(), busi.TransOutUID, -30, "") |
|||
}) |
|||
}) |
|||
assert.Nil(t, err) |
|||
waitTransProcessed(msg.Gid) |
|||
assert.Equal(t, []string{StatusSucceed}, getBranchesStatus(msg.Gid)) |
|||
assert.Equal(t, StatusSucceed, getTransStatus(msg.Gid)) |
|||
assertNotSameBalance(t, before, "mongo") |
|||
} |
|||
|
|||
func TestMsgMongoDoBusiFailed(t *testing.T) { |
|||
before := getBeforeBalances("mongo") |
|||
gid := dtmimp.GetFuncName() |
|||
req := busi.GenTransReq(30, false, false) |
|||
msg := dtmcli.NewMsg(DtmServer, gid). |
|||
Add(busi.Busi+"/SagaMongoTransIn", req) |
|||
err := msg.DoAndSubmit(Busi+"/MongoQueryPrepared", func(bb *dtmcli.BranchBarrier) error { |
|||
return errors.New("an error") |
|||
}) |
|||
assert.Error(t, err) |
|||
assertSameBalance(t, before, "mongo") |
|||
} |
|||
|
|||
func TestMsgMongoDoCommitFailed(t *testing.T) { |
|||
before := getBeforeBalances("mongo") |
|||
gid := dtmimp.GetFuncName() |
|||
req := busi.GenTransReq(30, false, false) |
|||
msg := dtmcli.NewMsg(DtmServer, gid). |
|||
Add(busi.Busi+"/SagaMongoTransIn", req) |
|||
err := msg.DoAndSubmit(Busi+"/MongoQueryPrepared", func(bb *dtmcli.BranchBarrier) error { |
|||
return bb.MongoCall(busi.MongoGet(), func(sc mongo.SessionContext) error { |
|||
err := busi.SagaMongoAdjustBalance(sc, sc.Client(), busi.TransOutUID, -30, "") |
|||
assert.Nil(t, err) |
|||
return errors.New("commit failed") |
|||
}) |
|||
}) |
|||
assert.Error(t, err) |
|||
assertSameBalance(t, before, "mongo") |
|||
} |
|||
|
|||
func TestMsgMongoDoCommitAfterFailed(t *testing.T) { |
|||
before := getBeforeBalances("mongo") |
|||
gid := dtmimp.GetFuncName() |
|||
req := busi.GenTransReq(30, false, false) |
|||
msg := dtmcli.NewMsg(DtmServer, gid). |
|||
Add(busi.Busi+"/SagaMongoTransIn", req) |
|||
err := msg.DoAndSubmit(Busi+"/MongoQueryPrepared", func(bb *dtmcli.BranchBarrier) error { |
|||
err := bb.MongoCall(busi.MongoGet(), func(sc mongo.SessionContext) error { |
|||
return busi.SagaMongoAdjustBalance(sc, sc.Client(), busi.TransOutUID, -30, "") |
|||
}) |
|||
assert.Nil(t, err) |
|||
return errors.New("an error") |
|||
}) |
|||
assert.Error(t, err) |
|||
waitTransProcessed(gid) |
|||
assertNotSameBalance(t, before, "mongo") |
|||
} |
|||
@ -0,0 +1,46 @@ |
|||
/* |
|||
* Copyright (c) 2021 yedf. All rights reserved. |
|||
* Use of this source code is governed by a BSD-style |
|||
* license that can be found in the LICENSE file. |
|||
*/ |
|||
|
|||
package test |
|||
|
|||
import ( |
|||
"testing" |
|||
|
|||
"github.com/dtm-labs/dtm/dtmcli" |
|||
"github.com/dtm-labs/dtm/dtmcli/dtmimp" |
|||
"github.com/dtm-labs/dtm/test/busi" |
|||
"github.com/stretchr/testify/assert" |
|||
) |
|||
|
|||
func TestSagaBarrierMongoNormal(t *testing.T) { |
|||
before := getBeforeBalances("mongo") |
|||
saga := genSagaBarrierMongo(dtmimp.GetFuncName(), false) |
|||
err := saga.Submit() |
|||
assert.Nil(t, err) |
|||
waitTransProcessed(saga.Gid) |
|||
assert.Equal(t, []string{StatusPrepared, StatusSucceed, StatusPrepared, StatusSucceed}, getBranchesStatus(saga.Gid)) |
|||
assert.Equal(t, StatusSucceed, getTransStatus(saga.Gid)) |
|||
assertNotSameBalance(t, before, "mongo") |
|||
} |
|||
|
|||
func TestSagaBarrierMongoRollback(t *testing.T) { |
|||
before := getBeforeBalances("mongo") |
|||
saga := genSagaBarrierMongo(dtmimp.GetFuncName(), true) |
|||
err := saga.Submit() |
|||
assert.Nil(t, err) |
|||
waitTransProcessed(saga.Gid) |
|||
assert.Equal(t, StatusFailed, getTransStatus(saga.Gid)) |
|||
assert.Equal(t, []string{StatusSucceed, StatusSucceed, StatusSucceed, StatusFailed}, getBranchesStatus(saga.Gid)) |
|||
assertSameBalance(t, before, "mongo") |
|||
} |
|||
|
|||
func genSagaBarrierMongo(gid string, transInFailed bool) *dtmcli.Saga { |
|||
req := busi.GenTransReq(30, false, transInFailed) |
|||
req.Store = "mongo" |
|||
return dtmcli.NewSaga(DtmServer, gid). |
|||
Add(Busi+"/SagaMongoTransOut", Busi+"/SagaMongoTransOutCom", req). |
|||
Add(Busi+"/SagaMongoTransIn", Busi+"/SagaMongoTransInCom", req) |
|||
} |
|||
Loading…
Reference in new issue