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

add comment

Browse Source
This commit is contained in:
yedongfu 2021-07-16 12:34:23 +08:00
parent a3551fdfb4
commit 90d84b77a2
10 changed files with 72 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/data.go
View File

@ -8,6 +8,7 @@ import (
"github.com/yedf/dtm/common"
)
// RunSqlScript 1
func RunSqlScript(mysql map[string]string, script string) {
conf := map[string]string{}
common.MustRemarshal(mysql, &conf)
@ -29,6 +30,7 @@ func RunSqlScript(mysql map[string]string, script string) {
}
}
// PopulateMysql populate example mysql data
func PopulateMysql() {
common.InitApp(common.GetProjectDir(), &Config)
Config.Mysql["database"] = dbName

27
examples/main_base.go
View File

@ -10,12 +10,16 @@ import (
)
const (
BusiApi = "/api/busi"
// BusiAPI busi api prefix
BusiAPI = "/api/busi"
// BusiPort busi server port
BusiPort = 8081
)
var Busi string = fmt.Sprintf("http://localhost:%d%s", BusiPort, BusiApi)
// Busi busi service url prefix
var Busi string = fmt.Sprintf("http://localhost:%d%s", BusiPort, BusiAPI)
// BaseAppStartup base app startup
func BaseAppStartup() *gin.Engine {
logrus.Printf("examples starting")
app := common.GetGinApp()
@ -26,14 +30,17 @@ func BaseAppStartup() *gin.Engine {
return app
}
// AutoEmptyString auto reset to empty when used once
type AutoEmptyString struct {
value string
}
// SetOnce set a value once
func (s *AutoEmptyString) SetOnce(v string) {
s.value = v
}
// Fetch fetch the stored value, then reset the value to empty
func (s *AutoEmptyString) Fetch() string {
v := s.value
s.value = ""
@ -50,6 +57,7 @@ type mainSwitchType struct {
CanSubmitResult AutoEmptyString
}
// MainSwitch controls busi success or fail
var MainSwitch mainSwitchType
func handleGeneralBusiness(c *gin.Context, result1 string, result2 string, busi string) (interface{}, error) {
@ -60,26 +68,27 @@ func handleGeneralBusiness(c *gin.Context, result1 string, result2 string, busi
}
// BaseAddRoute add base route handler
func BaseAddRoute(app *gin.Engine) {
app.POST(BusiApi+"/TransIn", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.POST(BusiAPI+"/TransIn", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
return handleGeneralBusiness(c, MainSwitch.TransInResult.Fetch(), reqFrom(c).TransInResult, "transIn")
}))
app.POST(BusiApi+"/TransOut", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.POST(BusiAPI+"/TransOut", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
return handleGeneralBusiness(c, MainSwitch.TransOutResult.Fetch(), reqFrom(c).TransOutResult, "transIn")
}))
app.POST(BusiApi+"/TransInConfirm", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.POST(BusiAPI+"/TransInConfirm", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
return handleGeneralBusiness(c, MainSwitch.TransInConfirmResult.Fetch(), "", "transIn")
}))
app.POST(BusiApi+"/TransOutConfirm", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.POST(BusiAPI+"/TransOutConfirm", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
return handleGeneralBusiness(c, MainSwitch.TransOutConfirmResult.Fetch(), "", "transIn")
}))
app.POST(BusiApi+"/TransInRevert", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.POST(BusiAPI+"/TransInRevert", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
return handleGeneralBusiness(c, MainSwitch.TransInRevertResult.Fetch(), "", "transIn")
}))
app.POST(BusiApi+"/TransOutRevert", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.POST(BusiAPI+"/TransOutRevert", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
return handleGeneralBusiness(c, MainSwitch.TransOutRevertResult.Fetch(), "", "transIn")
}))
app.GET(BusiApi+"/CanSubmit", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.GET(BusiAPI+"/CanSubmit", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
logrus.Printf("%s CanSubmit", c.Query("gid"))
return common.OrString(MainSwitch.CanSubmitResult.Fetch(), "SUCCESS"), nil
}))

2
examples/main_msg.go
View File

@ -6,9 +6,11 @@ import (
"github.com/yedf/dtm/dtmcli"
)
// MsgSetup 1
func MsgSetup(app *gin.Engine) {
}
// MsgFireRequest 1
func MsgFireRequest() {
logrus.Printf("a busi transaction begin")
req := &TransReq{

2
examples/main_saga.go
View File

@ -6,9 +6,11 @@ import (
"github.com/yedf/dtm/dtmcli"
)
// SagaSetup 1
func SagaSetup(app *gin.Engine) {
}
// SagaFireRequest 1
func SagaFireRequest() {
logrus.Printf("a saga busi transaction begin")
req := &TransReq{

12
examples/main_saga_barrier.go
View File

@ -10,6 +10,7 @@ import (
"gorm.io/gorm"
)
// SagaBarrierFireRequest 1
func SagaBarrierFireRequest() {
logrus.Printf("a busi transaction begin")
req := &TransReq{Amount: 30}
@ -21,13 +22,12 @@ func SagaBarrierFireRequest() {
e2p(err)
}
// api
// SagaBarrierAddRoute 1
func SagaBarrierAddRoute(app *gin.Engine) {
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))
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)
}

4
examples/main_tcc.go
View File

@ -7,8 +7,9 @@ import (
"github.com/yedf/dtm/dtmcli"
)
// TccSetup 1
func TccSetup(app *gin.Engine) {
app.POST(BusiApi+"/TransInTcc", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.POST(BusiAPI+"/TransInTcc", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
tcc, err := dtmcli.TccFromReq(c)
if err != nil {
return nil, err
@ -25,6 +26,7 @@ func TccSetup(app *gin.Engine) {
}))
}
// TccFireRequest 1
func TccFireRequest() {
logrus.Printf("tcc transaction begin")
_, err := dtmcli.TccGlobalTransaction(DtmServer, func(tcc *dtmcli.Tcc) (rerr error) {

32
examples/main_tcc_barrier.go
View File

@ -10,6 +10,7 @@ import (
"github.com/yedf/dtm/dtmcli"
)
// TccBarrierFireRequest 1
func TccBarrierFireRequest() {
logrus.Printf("tcc transaction begin")
_, err := dtmcli.TccGlobalTransaction(DtmServer, func(tcc *dtmcli.Tcc) (rerr error) {
@ -33,20 +34,19 @@ func TccBarrierFireRequest() {
e2p(err)
}
// api
// TccBarrierAddRoute 1
func TccBarrierAddRoute(app *gin.Engine) {
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))
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
const transOutUid = 2
const transInUID = 1
const transOutUID = 2
func adjustTrading(sdb *sql.DB, uid int, amount int) (interface{}, error) {
db := common.SQLDB2DB(sdb)
@ -75,36 +75,36 @@ func adjustBalance(sdb *sql.DB, uid int, amount int) (interface{}, error) {
// TCC下，转入
func tccBarrierTransInTry(c *gin.Context) (interface{}, error) {
return dtmcli.ThroughBarrierCall(dbGet().ToSQLDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustTrading(sdb, transInUid, reqFrom(c).Amount)
return adjustTrading(sdb, transInUID, reqFrom(c).Amount)
})
}
func tccBarrierTransInConfirm(c *gin.Context) (interface{}, error) {
return dtmcli.ThroughBarrierCall(dbGet().ToSQLDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustBalance(sdb, transInUid, reqFrom(c).Amount)
return adjustBalance(sdb, transInUID, reqFrom(c).Amount)
})
}
func tccBarrierTransInCancel(c *gin.Context) (interface{}, error) {
return dtmcli.ThroughBarrierCall(dbGet().ToSQLDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustTrading(sdb, transInUid, -reqFrom(c).Amount)
return adjustTrading(sdb, transInUID, -reqFrom(c).Amount)
})
}
func tccBarrierTransOutTry(c *gin.Context) (interface{}, error) {
return dtmcli.ThroughBarrierCall(dbGet().ToSQLDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustTrading(sdb, transOutUid, -reqFrom(c).Amount)
return adjustTrading(sdb, transOutUID, -reqFrom(c).Amount)
})
}
func tccBarrierTransOutConfirm(c *gin.Context) (interface{}, error) {
return dtmcli.ThroughBarrierCall(dbGet().ToSQLDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustBalance(sdb, transOutUid, -reqFrom(c).Amount)
return adjustBalance(sdb, transOutUID, -reqFrom(c).Amount)
})
}
func tccBarrierTransOutCancel(c *gin.Context) (interface{}, error) {
return dtmcli.ThroughBarrierCall(dbGet().ToSQLDB(), dtmcli.TransInfoFromReq(c), func(sdb *sql.DB) (interface{}, error) {
return adjustTrading(sdb, transOutUid, reqFrom(c).Amount)
return adjustTrading(sdb, transOutUID, reqFrom(c).Amount)
})
}

13
examples/main_xa.go
View File

@ -9,28 +9,34 @@ import (
"github.com/yedf/dtm/dtmcli"
)
// XaClient XA client connection
var XaClient *dtmcli.XaClient = nil
// UserAccount busi model
type UserAccount struct {
common.ModelBase
UserId int
Balance string
}
// TableName gorm table name
func (u *UserAccount) TableName() string { return "user_account" }
// UserAccountTrading freeze user account table
type UserAccountTrading struct {
common.ModelBase
UserId int
TradingBalance string
}
// TableName gorm table name
func (u *UserAccountTrading) TableName() string { return "user_account_trading" }
func dbGet() *common.DB {
return common.DbGet(Config.Mysql)
}
// XaFireRequest 1
func XaFireRequest() {
_, err := XaClient.XaGlobalTransaction(func(xa *dtmcli.Xa) (rerr error) {
defer common.P2E(&rerr)
@ -44,10 +50,10 @@ func XaFireRequest() {
e2p(err)
}
// api
// XaSetup 1
func XaSetup(app *gin.Engine) {
app.POST(BusiApi+"/TransInXa", common.WrapHandler(xaTransIn))
app.POST(BusiApi+"/TransOutXa", common.WrapHandler(xaTransOut))
app.POST(BusiAPI+"/TransInXa", common.WrapHandler(xaTransIn))
app.POST(BusiAPI+"/TransOutXa", common.WrapHandler(xaTransOut))
Config.Mysql["database"] = "dtm_busi"
XaClient = dtmcli.NewXaClient(DtmServer, Config.Mysql, app, Busi+"/xa")
}
@ -80,6 +86,7 @@ func xaTransOut(c *gin.Context) (interface{}, error) {
return M{"result": "SUCCESS"}, nil
}
// ResetXaData 1
func ResetXaData() {
db := dbGet()
db.Must().Exec("truncate user_account")

14
examples/quick_start.go
View File

@ -13,12 +13,12 @@ import (
// 启动命令：go run app/main.go qs
// 事务参与者的服务地址
const qsBusiApi = "/api/busi_start"
const qsBusiAPI = "/api/busi_start"
const qsBusiPort = 8082
var qsBusi = fmt.Sprintf("http://localhost:%d%s", qsBusiPort, qsBusiApi)
var qsBusi = fmt.Sprintf("http://localhost:%d%s", qsBusiPort, qsBusiAPI)
// 被app/main.go调用，启动服务并运行示例
// QuickStarMain called by app/main.go
func QuickStarMain() {
qsStartSvr()
qsFireRequest()
@ -46,16 +46,16 @@ func qsFireRequest() {
}
func qsAddRoute(app *gin.Engine) {
app.POST(qsBusiApi+"/TransIn", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.POST(qsBusiAPI+"/TransIn", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
return M{"result": "SUCCESS"}, nil
}))
app.POST(qsBusiApi+"/TransInCompensate", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.POST(qsBusiAPI+"/TransInCompensate", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
return M{"result": "SUCCESS"}, nil
}))
app.POST(qsBusiApi+"/TransOut", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.POST(qsBusiAPI+"/TransOut", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
return M{"result": "SUCCESS"}, nil
}))
app.POST(qsBusiApi+"/TransOutCompensate", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
app.POST(qsBusiAPI+"/TransOutCompensate", common.WrapHandler(func(c *gin.Context) (interface{}, error) {
return M{"result": "SUCCESS"}, nil
}))
}

7
examples/types.go
View File

@ -10,16 +10,20 @@ import (
var e2p = common.E2P
// M alias
type M = map[string]interface{}
// 指定dtm服务地址
// DtmServer dtm service address
const DtmServer = "http://localhost:8080/api/dtmsvr"
const (
// SagaBarrierBusiPort saga barrier sample port
SagaBarrierBusiPort = iota + 8090
// TccBarrierBusiPort tcc barrier sample port
TccBarrierBusiPort
)
// TransReq transaction request payload
type TransReq struct {
Amount int `json:"amount"`
TransInResult string `json:"transInResult"`
@ -30,6 +34,7 @@ func (t *TransReq) String() string {
return fmt.Sprintf("amount: %d transIn: %s transOut: %s", t.Amount, t.TransInResult, t.TransOutResult)
}
// GenTransReq 1
func GenTransReq(amount int, outFailed bool, inFailed bool) *TransReq {
return &TransReq{
Amount: amount,