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

refactor main

Browse Source
This commit is contained in:
yedongfu 2021-07-10 22:54:02 +08:00
parent 91b1ccdc46
commit 7aabd05871
10 changed files with 61 additions and 169 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
app/main.go
View File

@ -12,54 +12,48 @@ import (
type M = map[string]interface{}
func wait() {
for {
time.Sleep(10000 * time.Second)
}
}
func main() {
if len(os.Args) > 1 && (os.Args[1] == "quick_start" || os.Args[1] == "qs") {
dtmsvr.PopulateMysql()
if len(os.Args) == 1 || os.Args[1] == "dtmsvr" { // 只启动dtmsvr
dtmsvr.MainStart()
examples.StartMain()
wait()
}
app := examples.BaseAppNew()
examples.BaseAppSetup(app)
if len(os.Args) == 1 || os.Args[1] == "dtmsvr" { // 只启动dtmsvr
go dtmsvr.MainStart()
} else if os.Args[1] == "xa" { // 启动xa示例
// 下面都是运行示例，因此首先把服务器的数据重新准备好
dtmsvr.PopulateMysql()
dtmsvr.MainStart()
// quick_start 比较独立，单独作为一个例子运行，方便新人上手
if len(os.Args) > 1 && (os.Args[1] == "quick_start" || os.Args[1] == "qs") {
examples.QuickStarMain()
wait()
}
// 下面是各类的例子
examples.PopulateMysql()
app := examples.BaseAppStartup()
if os.Args[1] == "xa" { // 启动xa示例
examples.XaSetup(app)
examples.BaseAppStart(app)
examples.XaFireRequest()
} else if os.Args[1] == "saga" { // 启动saga示例
dtmsvr.PopulateMysql()
dtmsvr.MainStart()
examples.SagaSetup(app)
examples.BaseAppStart(app)
examples.SagaFireRequest()
} else if os.Args[1] == "all" { // 运行所有示例
dtmsvr.PopulateMysql()
examples.PopulateMysql()
dtmsvr.MainStart()
examples.SagaSetup(app)
examples.TccSetup(app)
examples.XaSetup(app)
examples.BaseAppStart(app)
examples.SagaFireRequest()
examples.TccFireRequest()
examples.XaFireRequest()
} else if os.Args[1] == "saga_barrier" {
dtmsvr.PopulateMysql()
dtmsvr.MainStart()
examples.PopulateMysql()
examples.SagaBarrierMainStart()
examples.SagaBarrierAddRoute(app)
examples.SagaBarrierFireRequest()
} else if os.Args[1] == "tcc_barrier" {
dtmsvr.PopulateMysql()
dtmsvr.MainStart()
examples.PopulateMysql()
examples.TccBarrierMainStart()
examples.TccBarrierAddRoute(app)
examples.TccBarrierFireRequest()
} else {
logrus.Fatalf("unknown arg: %s", os.Args[1])
}

4
dtmsvr/dtmsvr_test.go
View File

@ -33,13 +33,11 @@ func TestDtmSvr(t *testing.T) {
TransProcessedTestChan = make(chan string, 1)
// 启动组件
go StartSvr()
app := examples.BaseAppNew()
examples.BaseAppSetup(app)
app := examples.BaseAppStartup()
examples.SagaSetup(app)
examples.TccSetup(app)
examples.XaSetup(app)
examples.MsgSetup(app)
examples.BaseAppStart(app)
// 清理数据
e2p(dbGet().Exec("truncate trans_global").Error)

10
examples/main_base.go
View File

@ -16,16 +16,14 @@ const (
var Busi string = fmt.Sprintf("http://localhost:%d%s", BusiPort, BusiApi)
func BaseAppNew() *gin.Engine {
func BaseAppStartup() *gin.Engine {
logrus.Printf("examples starting")
app := common.GetGinApp()
return app
}
func BaseAppStart(app *gin.Engine) {
BaseAddRoute(app)
logrus.Printf("Starting busi at: %d", BusiPort)
go app.Run(fmt.Sprintf(":%d", BusiPort))
time.Sleep(100 * time.Millisecond)
return app
}
type AutoEmptyString struct {
@ -62,7 +60,7 @@ func handleGeneralBusiness(c *gin.Context, result1 string, result2 string, busi
}
func BaseAppSetup(app *gin.Engine) {
func BaseAddRoute(app *gin.Engine) {
app.POST(BusiApi+"/TransIn", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
return handleGeneralBusiness(c, MainSwitch.TransInResult.Fetch(), reqFrom(c).TransInResult, "transIn")
}))

10
examples/main_msg.go
View File

@ -1,21 +1,11 @@
package examples
import (
"time"
"github.com/gin-gonic/gin"
"github.com/sirupsen/logrus"
"github.com/yedf/dtm/dtmcli"
)
func MsgMain() {
app := BaseAppNew()
BaseAppSetup(app)
BaseAppStart(app)
MsgFireRequest()
time.Sleep(1000 * time.Second)
}
func MsgSetup(app *gin.Engine) {
}

12
examples/main_saga.go
View File

@ -1,23 +1,11 @@
package examples
import (
"time"
"github.com/gin-gonic/gin"
"github.com/sirupsen/logrus"
"github.com/yedf/dtm/dtmcli"
)
func SagaMain() {
app := BaseAppNew()
BaseAppSetup(app)
TccSetup(app)
go BaseAppStart(app)
time.Sleep(100 * time.Millisecond)
TccFireRequest()
time.Sleep(1000 * time.Second)
}
func SagaSetup(app *gin.Engine) {
}

65
examples/main_saga_barrier.go
View File

@ -2,8 +2,6 @@ package examples
import (
"database/sql"
"fmt"
"time"
"github.com/gin-gonic/gin"
"github.com/sirupsen/logrus"
@ -12,30 +10,12 @@ import (
"gorm.io/gorm"
)
// 事务参与者的服务地址
const SagaBarrierBusiApi = "/api/busi_saga_barrier"
var SagaBarrierBusi = fmt.Sprintf("http://localhost:%d%s", SagaBarrierBusiPort, SagaBarrierBusiApi)
func SagaBarrierMainStart() {
SagaBarrierStartSvr()
SagaBarrierFireRequest()
}
func SagaBarrierStartSvr() {
logrus.Printf("saga barrier examples starting")
app := common.GetGinApp()
SagaBarrierAddRoute(app)
go app.Run(fmt.Sprintf(":%d", SagaBarrierBusiPort))
time.Sleep(100 * time.Millisecond)
}
func SagaBarrierFireRequest() {
logrus.Printf("a busi transaction begin")
req := &TransReq{Amount: 30}
saga := dtmcli.NewSaga(DtmServer).
Add(SagaBarrierBusi+"/TransOut", SagaBarrierBusi+"/TransOutCompensate", req).
Add(SagaBarrierBusi+"/TransIn", SagaBarrierBusi+"/TransInCompensate", req)
Add(Busi+"/SagaBTransOut", Busi+"/SagaBTransOutCompensate", req).
Add(Busi+"/SagaBTransIn", Busi+"/SagaBTransInCompensate", req)
logrus.Printf("busi trans submit")
err := saga.Submit()
e2p(err)
@ -44,49 +24,40 @@ func SagaBarrierFireRequest() {
// api
func SagaBarrierAddRoute(app *gin.Engine) {
app.POST(SagaBarrierBusiApi+"/TransIn", common.WrapHandler(sagaBarrierTransIn))
app.POST(SagaBarrierBusiApi+"/TransInCompensate", common.WrapHandler(sagaBarrierTransInCompensate))
app.POST(SagaBarrierBusiApi+"/TransOut", common.WrapHandler(sagaBarrierTransOut))
app.POST(SagaBarrierBusiApi+"/TransOutCompensate", common.WrapHandler(sagaBarrierTransOutCompensate))
logrus.Printf("examples listening at %d", SagaBarrierBusiPort)
app.POST(BusiApi+"/SagaBTransIn", common.WrapHandler(sagaBarrierTransIn))
app.POST(BusiApi+"/SagaBTransInCompensate", common.WrapHandler(sagaBarrierTransInCompensate))
app.POST(BusiApi+"/SagaBTransOut", common.WrapHandler(sagaBarrierTransOut))
app.POST(BusiApi+"/SagaBTransOutCompensate", common.WrapHandler(sagaBarrierTransOutCompensate))
logrus.Printf("examples listening at %d", BusiPort)
}
func sagaBarrierAdjustBalance(sdb *sql.DB, uid int, amount int) (interface{}, error) {
db := common.SqlDB2DB(sdb)
dbr := db.Model(&UserAccount{}).Where("user_id = ?", 1).Update("balance", gorm.Expr("balance + ?", amount))
return "SUCCESS", dbr.Error
}
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 = ?", 1).
Update("balance", gorm.Expr("balance + ?", req.Amount))
return "SUCCESS", dbr.Error
return sagaBarrierAdjustBalance(sdb, 1, reqFrom(c).Amount)
})
}
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 = ?", 1).
Update("balance", gorm.Expr("balance - ?", req.Amount))
return "SUCCESS", dbr.Error
return sagaBarrierAdjustBalance(sdb, 1, -reqFrom(c).Amount)
})
}
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 = ?", 2).
Update("balance", gorm.Expr("balance - ?", req.Amount))
return "SUCCESS", dbr.Error
return sagaBarrierAdjustBalance(sdb, 2, -reqFrom(c).Amount)
})
}
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 = ?", 2).
Update("balance", gorm.Expr("balance + ?", req.Amount))
return "SUCCESS", dbr.Error
return sagaBarrierAdjustBalance(sdb, 2, reqFrom(c).Amount)
})
}

12
examples/main_tcc.go
View File

@ -1,24 +1,12 @@
package examples
import (
"time"
"github.com/gin-gonic/gin"
"github.com/sirupsen/logrus"
"github.com/yedf/dtm/common"
"github.com/yedf/dtm/dtmcli"
)
func TccMain() {
app := BaseAppNew()
BaseAppSetup(app)
TccSetup(app)
go BaseAppStart(app)
time.Sleep(100 * time.Millisecond)
TccFireRequest()
time.Sleep(1000 * time.Second)
}
func TccSetup(app *gin.Engine) {
app.POST(BusiApi+"/TransInTcc", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
tcc, err := dtmcli.TccFromReq(c)

55
examples/main_tcc_barrier.go
View File

@ -3,7 +3,6 @@ package examples
import (
"database/sql"
"fmt"
"time"
"github.com/gin-gonic/gin"
"github.com/sirupsen/logrus"
@ -11,35 +10,17 @@ import (
"github.com/yedf/dtm/dtmcli"
)
// 事务参与者的服务地址
const TccBarrierBusiApi = "/api/busi_saga_barrier"
var TccBarrierBusi = fmt.Sprintf("http://localhost:%d%s", TccBarrierBusiPort, TccBarrierBusiApi)
func TccBarrierMainStart() {
TccBarrierStartSvr()
TccBarrierFireRequest()
}
func TccBarrierStartSvr() {
logrus.Printf("saga barrier examples starting")
app := common.GetGinApp()
TccBarrierAddRoute(app)
go app.Run(fmt.Sprintf(":%d", TccBarrierBusiPort))
time.Sleep(100 * time.Millisecond)
}
func TccBarrierFireRequest() {
logrus.Printf("tcc transaction begin")
_, err := dtmcli.TccGlobalTransaction(DtmServer, func(tcc *dtmcli.Tcc) (rerr error) {
res1, rerr := tcc.CallBranch(&TransReq{Amount: 30}, TccBarrierBusi+"/TransOutTry", TccBarrierBusi+"/TransOutConfirm", TccBarrierBusi+"/TransOutRevert")
res1, rerr := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TccBTransOutTry", Busi+"/TccBTransOutConfirm", Busi+"/TccBTransOutRevert")
if rerr != nil {
return
}
if res1.StatusCode() != 200 {
return fmt.Errorf("bad status code: %d", res1.StatusCode())
}
res2, rerr := tcc.CallBranch(&TransReq{Amount: 30}, TccBarrierBusi+"/TransInTry", TccBarrierBusi+"/TransInConfirm", TccBarrierBusi+"/TransInRevert")
res2, rerr := tcc.CallBranch(&TransReq{Amount: 30}, Busi+"/TccBTransInTry", Busi+"/TccBTransInConfirm", Busi+"/TccBTransInRevert")
if rerr != nil {
return
}
@ -55,13 +36,13 @@ func TccBarrierFireRequest() {
// api
func TccBarrierAddRoute(app *gin.Engine) {
app.POST(TccBarrierBusiApi+"/TransInTry", common.WrapHandler(tccBarrierTransInTry))
app.POST(TccBarrierBusiApi+"/TransInConfirm", common.WrapHandler(tccBarrierTransInConfirm))
app.POST(TccBarrierBusiApi+"/TransInCancel", common.WrapHandler(tccBarrierTransInCancel))
app.POST(TccBarrierBusiApi+"/TransOutTry", common.WrapHandler(tccBarrierTransOutTry))
app.POST(TccBarrierBusiApi+"/TransOutConfirm", common.WrapHandler(tccBarrierTransOutConfirm))
app.POST(TccBarrierBusiApi+"/TransOutCancel", common.WrapHandler(tccBarrierTransOutCancel))
logrus.Printf("examples listening at %d", TccBarrierBusiPort)
app.POST(BusiApi+"/TccBTransInTry", common.WrapHandler(tccBarrierTransInTry))
app.POST(BusiApi+"/TccBTransInConfirm", common.WrapHandler(tccBarrierTransInConfirm))
app.POST(BusiApi+"/TccBTransInCancel", common.WrapHandler(tccBarrierTransInCancel))
app.POST(BusiApi+"/TccBTransOutTry", common.WrapHandler(tccBarrierTransOutTry))
app.POST(BusiApi+"/TccBTransOutConfirm", common.WrapHandler(tccBarrierTransOutConfirm))
app.POST(BusiApi+"/TccBTransOutCancel", common.WrapHandler(tccBarrierTransOutCancel))
logrus.Printf("examples listening at %d", BusiPort)
}
const transInUid = 1
@ -93,43 +74,37 @@ func adjustBalance(sdb *sql.DB, uid int, amount int) (interface{}, error) {
// TCC下，转入
func tccBarrierTransInTry(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
return dtmcli.ThroughBarrierCall(dbGet().ToSqlDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustTrading(sdb, transInUid, req.Amount)
return adjustTrading(sdb, transInUid, reqFrom(c).Amount)
})
}
func tccBarrierTransInConfirm(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
return dtmcli.ThroughBarrierCall(dbGet().ToSqlDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustBalance(sdb, transInUid, req.Amount)
return adjustBalance(sdb, transInUid, reqFrom(c).Amount)
})
}
func tccBarrierTransInCancel(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
return dtmcli.ThroughBarrierCall(dbGet().ToSqlDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustTrading(sdb, transInUid, -req.Amount)
return adjustTrading(sdb, transInUid, -reqFrom(c).Amount)
})
}
func tccBarrierTransOutTry(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
return dtmcli.ThroughBarrierCall(dbGet().ToSqlDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustTrading(sdb, transOutUid, -req.Amount)
return adjustTrading(sdb, transOutUid, -reqFrom(c).Amount)
})
}
func tccBarrierTransOutConfirm(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
return dtmcli.ThroughBarrierCall(dbGet().ToSqlDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustBalance(sdb, transOutUid, -req.Amount)
return adjustBalance(sdb, transOutUid, -reqFrom(c).Amount)
})
}
func tccBarrierTransOutCancel(c *gin.Context) (interface{}, error) {
req := reqFrom(c)
return dtmcli.ThroughBarrierCall(dbGet().ToSqlDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustTrading(sdb, transOutUid, req.Amount)
return adjustTrading(sdb, transOutUid, reqFrom(c).Amount)
})
}

10
examples/main_xa.go
View File

@ -2,7 +2,6 @@ package examples
import (
"fmt"
"time"
"github.com/gin-gonic/gin"
"github.com/yedf/dtm/common"
@ -32,15 +31,6 @@ func dbGet() *common.DB {
return common.DbGet(Config.Mysql)
}
func XaMain() {
app := BaseAppNew()
XaSetup(app)
go BaseAppStart(app)
time.Sleep(100 * time.Millisecond)
XaFireRequest()
time.Sleep(1000 * time.Second)
}
func XaFireRequest() {
_, err := XaClient.XaGlobalTransaction(func(gid string) (rerr error) {
defer common.P2E(&rerr)

2
examples/quick_start.go
View File

@ -19,7 +19,7 @@ const qsBusiPort = 8082
var qsBusi = fmt.Sprintf("http://localhost:%d%s", qsBusiPort, qsBusiApi)
// 被app/main.go调用，启动服务并运行示例
func StartMain() {
func QuickStarMain() {
qsStartSvr()
qsFireRequest()
}