opensource/dtm
3
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

saga barrier seems ok

Browse Source
This commit is contained in:
yedongfu 2021-07-08 22:21:34 +08:00
parent aef42d59aa
commit b0102048a9
10 changed files with 50 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
app/main.go
View File

@ -11,36 +11,38 @@ import (
type M = map[string]interface{}
func wait() {
time.Sleep(10000 * time.Second)
}
func main() {
if len(os.Args) > 1 && (os.Args[1] == "quick_start" || os.Args[1] == "qs") {
dtmsvr.PopulateMysql()
dtmsvr.Main()
dtmsvr.MainStart()
examples.StartMain()
for {
time.Sleep(1000 * time.Second)
}
wait()
}
app := examples.BaseAppNew()
examples.BaseAppSetup(app)
if len(os.Args) == 1 || os.Args[1] == "saga" { // 默认情况下，展示saga例子
dtmsvr.PopulateMysql()
dtmsvr.Main()
dtmsvr.MainStart()
examples.SagaSetup(app)
examples.BaseAppStart(app)
examples.SagaFireRequest()
} else if os.Args[1] == "xa" { // 启动xa示例
dtmsvr.PopulateMysql()
dtmsvr.Main()
dtmsvr.MainStart()
examples.PopulateMysql()
examples.XaSetup(app)
examples.BaseAppStart(app)
examples.XaFireRequest()
} else if os.Args[1] == "dtmsvr" { // 只启动dtmsvr
go dtmsvr.StartSvr()
go dtmsvr.MainStart()
} else if os.Args[1] == "all" { // 运行所有示例
dtmsvr.PopulateMysql()
examples.PopulateMysql()
dtmsvr.Main()
dtmsvr.MainStart()
examples.SagaSetup(app)
examples.TccSetup(app)
examples.XaSetup(app)
@ -48,10 +50,13 @@ func main() {
examples.SagaFireRequest()
examples.TccFireRequest()
examples.XaFireRequest()
} else if os.Args[1] == "saga_barrier" {
dtmsvr.PopulateMysql()
dtmsvr.MainStart()
examples.PopulateMysql()
examples.SagaBarrierMainStart()
} else {
logrus.Fatalf("unknown arg: %s", os.Args[1])
}
for {
time.Sleep(1000 * time.Second)
}
wait()
}

1
common/types.go
View File

@ -114,6 +114,7 @@ func SqlDB2DB(sdb *sql.DB) *DB {
Conn: sdb,
}), &gorm.Config{})
E2P(err)
db.Use(&tracePlugin{})
return &DB{DB: db}
}

6
dtmcli/barrier.go
View File

@ -23,12 +23,16 @@ func (t *TransInfo) String() string {
}
func TransInfoFromReq(c *gin.Context) *TransInfo {
return &TransInfo{
ti := &TransInfo{
TransType: c.Query("trans_type"),
Gid: c.Query("gid"),
BranchID: c.Query("branch_id"),
BranchType: c.Query("branch_type"),
}
if ti.TransType == "" || ti.Gid == "" || ti.BranchID == "" || ti.BranchType == "" {
panic(fmt.Errorf("invlid trans info: %v", ti))
}
return ti
}
type BarrierModel struct {

8
dtmsvr/main.go
View File

@ -2,6 +2,7 @@ package dtmsvr
import (
"fmt"
"time"
"github.com/sirupsen/logrus"
"github.com/yedf/dtm/common"
@ -10,8 +11,8 @@ import (
var dtmsvrPort = 8080
func Main() {
go StartSvr()
func MainStart() {
StartSvr()
go CronCommitted()
go CronPrepared()
}
@ -23,7 +24,8 @@ func StartSvr() {
app := common.GetGinApp()
AddRoute(app)
logrus.Printf("dtmsvr listen at: %d", dtmsvrPort)
app.Run(fmt.Sprintf(":%d", dtmsvrPort))
go app.Run(fmt.Sprintf(":%d", dtmsvrPort))
time.Sleep(100 * time.Millisecond)
}
func PopulateMysql() {

9
dtmsvr/trans.go
View File

@ -117,6 +117,15 @@ func (trans *TransGlobal) Process(db *common.DB) {
trans.getProcessor().ProcessOnce(db, branches)
}
func (trans *TransGlobal) getBranchParams(branch *TransBranch) common.MS {
return common.MS{
"gid": trans.Gid,
"trans_type": trans.TransType,
"branch_id": branch.BranchID,
"branch_type": branch.BranchType,
}
}
func (t *TransGlobal) setNextCron(expireIn int64) []string {
t.NextCronInterval = expireIn
next := time.Now().Add(time.Duration(config.TransCronInterval) * time.Second)

2
dtmsvr/trans_msg.go
View File

@ -33,7 +33,7 @@ func (t *TransMsgProcessor) GenBranches() []TransBranch {
}
func (t *TransMsgProcessor) ExecBranch(db *common.DB, branch *TransBranch) {
resp, err := common.RestyClient.R().SetBody(branch.Data).SetQueryParam("gid", branch.Gid).Post(branch.Url)
resp, err := common.RestyClient.R().SetBody(branch.Data).SetQueryParams(t.getBranchParams(branch)).Post(branch.Url)
e2p(err)
body := resp.String()
if strings.Contains(body, "SUCCESS") {

2
dtmsvr/trans_saga.go
View File

@ -36,7 +36,7 @@ func (t *TransSagaProcessor) GenBranches() []TransBranch {
}
func (t *TransSagaProcessor) ExecBranch(db *common.DB, branch *TransBranch) {
resp, err := common.RestyClient.R().SetBody(branch.Data).SetQueryParam("gid", branch.Gid).Post(branch.Url)
resp, err := common.RestyClient.R().SetBody(branch.Data).SetQueryParams(t.getBranchParams(branch)).Post(branch.Url)
e2p(err)
body := resp.String()
if strings.Contains(body, "SUCCESS") {

2
dtmsvr/trans_tcc.go
View File

@ -20,7 +20,7 @@ func (t *TransTccProcessor) GenBranches() []TransBranch {
}
func (t *TransTccProcessor) ExecBranch(db *common.DB, branch *TransBranch) {
resp, err := common.RestyClient.R().SetBody(branch.Data).SetQueryParam("gid", branch.Gid).Post(branch.Url)
resp, err := common.RestyClient.R().SetBody(branch.Data).SetQueryParams(t.getBranchParams(branch)).Post(branch.Url)
e2p(err)
body := resp.String()
if strings.Contains(body, "SUCCESS") {

16
examples/main_saga_barrier.go
View File

@ -17,17 +17,17 @@ const SagaBarrierBusiApi = "/api/busi_saga_barrier"
var SagaBarrierBusi = fmt.Sprintf("http://localhost:%d%s", SagaBarrierBusiPort, SagaBarrierBusiApi)
func SagaBarrierMain() {
go SagaBarrierStartSvr()
func SagaBarrierMainStart() {
SagaBarrierStartSvr()
SagaBarrierFireRequest()
time.Sleep(1000 * time.Second)
}
func SagaBarrierStartSvr() {
logrus.Printf("saga barrier examples starting")
app := common.GetGinApp()
SagaBarrierAddRoute(app)
app.Run(fmt.Sprintf(":%d", SagaBarrierBusiPort))
go app.Run(fmt.Sprintf(":%d", SagaBarrierBusiPort))
time.Sleep(100 * time.Millisecond)
}
func SagaBarrierFireRequest() {
@ -55,7 +55,7 @@ func sagaBarrierTransIn(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
return dtmcli.ThroughBarrierCall(dbGet().ToSqlDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
db := common.SqlDB2DB(sdb)
dbr := db.Model(&UserAccount{}).Where("user_id = ?", c.Query("user_id")).
dbr := db.Model(&UserAccount{}).Where("user_id = ?", 1).
Update("balance", gorm.Expr("balance + ?", req.Amount))
return "SUCCESS", dbr.Error
})
@ -65,7 +65,7 @@ func sagaBarrierTransInCompensate(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
return dtmcli.ThroughBarrierCall(dbGet().ToSqlDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
db := common.SqlDB2DB(sdb)
dbr := db.Model(&UserAccount{}).Where("user_id = ?", c.Query("user_id")).
dbr := db.Model(&UserAccount{}).Where("user_id = ?", 1).
Update("balance", gorm.Expr("balance - ?", req.Amount))
return "SUCCESS", dbr.Error
})
@ -75,7 +75,7 @@ func sagaBarrierTransOut(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
return dtmcli.ThroughBarrierCall(dbGet().ToSqlDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
db := common.SqlDB2DB(sdb)
dbr := db.Model(&UserAccount{}).Where("user_id = ?", c.Query("user_id")).
dbr := db.Model(&UserAccount{}).Where("user_id = ?", 2).
Update("balance", gorm.Expr("balance - ?", req.Amount))
return "SUCCESS", dbr.Error
})
@ -85,7 +85,7 @@ func sagaBarrierTransOutCompensate(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
return dtmcli.ThroughBarrierCall(dbGet().ToSqlDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
db := common.SqlDB2DB(sdb)
dbr := db.Model(&UserAccount{}).Where("user_id = ?", c.Query("user_id")).
dbr := db.Model(&UserAccount{}).Where("user_id = ?", 2).
Update("balance", gorm.Expr("balance + ?", req.Amount))
return "SUCCESS", dbr.Error
})

6
examples/quick_start.go
View File

@ -20,16 +20,16 @@ var qsBusi = fmt.Sprintf("http://localhost:%d%s", qsBusiPort, qsBusiApi)
// 被app/main.go调用，启动服务并运行示例
func StartMain() {
go qsStartSvr()
qsStartSvr()
qsFireRequest()
time.Sleep(1000 * time.Second)
}
func qsStartSvr() {
app := common.GetGinApp()
qsAddRoute(app)
logrus.Printf("quick qs examples listening at %d", qsBusiPort)
app.Run(fmt.Sprintf(":%d", qsBusiPort))
go app.Run(fmt.Sprintf(":%d", qsBusiPort))
time.Sleep(100 * time.Millisecond)
}
func qsFireRequest() {