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

e2p

Browse Source
This commit is contained in:
yedongfu 2021-05-27 20:54:00 +08:00
parent 8a72a4d629
commit ff6b6e79b4
11 changed files with 47 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
common/types.go
View File

@ -93,7 +93,7 @@ func DbGet(conf map[string]string) *MyDb {
db1, err := gorm.Open(mysql.Open(dsn), &gorm.Config{
SkipDefaultTransaction: true,
})
PanicIfError(err)
E2P(err)
db1.Use(&tracePlugin{})
dbs[dsn] = &MyDb{DB: db1}
}
@ -114,11 +114,11 @@ func DbAlone(conf map[string]string) (*MyDb, *MyConn) {
dsn := GetDsn(conf)
logrus.Printf("opening alone mysql: %s", ReplaceDsnPassword(dsn))
mdb, err := sql.Open("mysql", dsn)
PanicIfError(err)
E2P(err)
gormDB, err := gorm.Open(mysql.New(mysql.Config{
Conn: mdb,
}), &gorm.Config{})
PanicIfError(err)
E2P(err)
gormDB.Use(&tracePlugin{})
return &MyDb{DB: gormDB}, &MyConn{Conn: mdb, Dsn: dsn}
}

20
common/utils.go
View File

@ -29,7 +29,7 @@ func OrString(ss ...string) string {
return ""
}
func Panic2Error(perr *error) {
func P2E(perr *error) {
if x := recover(); x != nil {
if e, ok := x.(error); ok {
*perr = e
@ -53,7 +53,7 @@ func init() {
gNode = node
}
func PanicIfError(err error) {
func E2P(err error) {
if err != nil {
panic(err)
}
@ -68,7 +68,7 @@ func If(condition bool, trueObj interface{}, falseObj interface{}) interface{} {
func MustMarshal(v interface{}) []byte {
b, err := json.Marshal(v)
PanicIfError(err)
E2P(err)
return b
}
@ -78,7 +78,7 @@ func MustMarshalString(v interface{}) string {
func MustUnmarshal(b []byte, obj interface{}) {
err := json.Unmarshal(b, obj)
PanicIfError(err)
E2P(err)
}
func MustUnmarshalString(s string, obj interface{}) {
MustUnmarshal([]byte(s), obj)
@ -86,9 +86,9 @@ func MustUnmarshalString(s string, obj interface{}) {
func MustRemarshal(from interface{}, to interface{}) {
b, err := json.Marshal(from)
PanicIfError(err)
E2P(err)
err = json.Unmarshal(b, to)
PanicIfError(err)
E2P(err)
}
func GetGinApp() *gin.Engine {
@ -130,7 +130,7 @@ func WrapHandler(fn func(*gin.Context) (interface{}, error)) gin.HandlerFunc {
c.Status(200)
c.Writer.Header().Add("Content-Type", "application/json")
_, err = c.Writer.Write(b)
PanicIfError(err)
E2P(err)
}
}
}
@ -154,7 +154,7 @@ func init() {
}
func CheckRestySuccess(resp *resty.Response, err error) {
PanicIfError(err)
E2P(err)
if !strings.Contains(resp.String(), "SUCCESS") {
panic(fmt.Errorf("resty response not success: %s", resp.String()))
}
@ -196,7 +196,7 @@ func InitApp(config interface{}) {
configLoaded[fileName] = true
viper.SetConfigFile(fileName)
err := viper.ReadInConfig()
PanicIfError(err)
E2P(err)
err = viper.Unmarshal(config)
PanicIfError(err)
E2P(err)
}

4
dtmsvr/api.go
View File

@ -44,7 +44,7 @@ func Rollback(c *gin.Context) (interface{}, error) {
func Branch(c *gin.Context) (interface{}, error) {
branch := TransBranchModel{}
err := c.BindJSON(&branch)
common.PanicIfError(err)
e2p(err)
db := dbGet()
db.Must().Clauses(clause.OnConflict{
DoNothing: true,
@ -55,7 +55,7 @@ func Branch(c *gin.Context) (interface{}, error) {
func getTransFromContext(c *gin.Context) *TransGlobalModel {
data := M{}
b, err := c.GetRawData()
common.PanicIfError(err)
e2p(err)
common.MustUnmarshal(b, &data)
logrus.Printf("creating trans model in prepare")
if data["trans_type"].(string) == "saga" {

2
dtmsvr/cron.go
View File

@ -21,7 +21,7 @@ func CronPreparedOnce(expire time.Duration) {
writeTransLog(sm.Gid, "saga touch prepared", "", "", "")
db.Must().Model(&sm).Update("id", sm.ID)
resp, err := common.RestyClient.R().SetQueryParam("gid", sm.Gid).Get(sm.QueryPrepared)
common.PanicIfError(err)
e2p(err)
body := resp.String()
if strings.Contains(body, "FAIL") {
preparedExpire := time.Now().Add(time.Duration(-config.PreparedExpire) * time.Second)

12
dtmsvr/dtmsvr_test.go
View File

@ -32,9 +32,9 @@ func TestDtmSvr(t *testing.T) {
time.Sleep(time.Duration(100 * 1000 * 1000))
// 清理数据
common.PanicIfError(dbGet().Exec("truncate trans_global").Error)
common.PanicIfError(dbGet().Exec("truncate trans_branch").Error)
common.PanicIfError(dbGet().Exec("truncate trans_log").Error)
e2p(dbGet().Exec("truncate trans_global").Error)
e2p(dbGet().Exec("truncate trans_branch").Error)
e2p(dbGet().Exec("truncate trans_log").Error)
examples.ResetXaData()
xaRollback(t)
@ -61,14 +61,14 @@ var initdb = dbGet()
func getSagaModel(gid string) *TransGlobalModel {
sm := TransGlobalModel{}
dbr := dbGet().Model(&sm).Where("gid=?", gid).First(&sm)
common.PanicIfError(dbr.Error)
e2p(dbr.Error)
return &sm
}
func getBranchesStatus(gid string) []string {
steps := []TransBranchModel{}
dbr := dbGet().Model(&TransBranchModel{}).Where("gid=?", gid).Find(&steps)
common.PanicIfError(dbr.Error)
e2p(dbr.Error)
status := []string{}
for _, step := range steps {
status = append(status, step.Status)
@ -93,7 +93,7 @@ func xaNormal(t *testing.T) {
common.CheckRestySuccess(resp, err)
return nil
})
common.PanicIfError(err)
e2p(err)
WaitTransProcessed(gid)
assert.Equal(t, []string{"finished", "finished"}, getBranchesStatus(gid))
}

2
dtmsvr/service.go
View File

@ -29,7 +29,7 @@ func saveCommitted(m *TransGlobalModel) {
}
return nil
})
common.PanicIfError(err)
e2p(err)
}
var TransProcessedTestChan chan string = nil // 用于测试时，通知处理结束

3
dtmsvr/types.go
View File

@ -10,6 +10,9 @@ import (
type M = map[string]interface{}
var p2e = common.P2E
var e2p = common.E2P
type TransGlobalModel struct {
common.ModelBase
Gid string `json:"gid"`

6
examples/saga_main.go
View File

@ -42,10 +42,10 @@ func sagaFireRequest() {
saga.Add(SagaBusi+"/TransOut", SagaBusi+"/TransOutCompensate", req)
saga.Add(SagaBusi+"/TransIn", SagaBusi+"/TransInCompensate", req)
err := saga.Prepare()
common.PanicIfError(err)
e2p(err)
logrus.Printf("busi trans commit")
err = saga.Commit()
common.PanicIfError(err)
e2p(err)
}
// api
@ -70,7 +70,7 @@ var TransQueryResult = ""
func transReqFromContext(c *gin.Context) *TransReq {
req := TransReq{}
err := c.BindJSON(&req)
common.PanicIfError(err)
e2p(err)
return &req
}

2
examples/types.go
View File

@ -2,6 +2,8 @@ package examples
import "github.com/yedf/dtm/common"
var e2p = common.E2P
type UserAccount struct {
common.ModelBase
UserId int

8
examples/xa_main.go
View File

@ -38,7 +38,7 @@ func XaStartSvr() {
func XaFireRequest() {
gid := common.GenGid()
err := XaClient.XaGlobalTransaction(gid, func() (rerr error) {
defer common.Panic2Error(&rerr)
defer common.P2E(&rerr)
req := GenTransReq(30, false, false)
resp, err := common.RestyClient.R().SetBody(req).SetQueryParams(map[string]string{
"gid": gid,
@ -52,7 +52,7 @@ func XaFireRequest() {
common.CheckRestySuccess(resp, err)
return nil
})
common.PanicIfError(err)
e2p(err)
}
// api
@ -71,7 +71,7 @@ func XaTransIn(c *gin.Context) (interface{}, error) {
Update("balance", gorm.Expr("balance - ?", req.Amount))
return dbr.Error
})
common.PanicIfError(err)
e2p(err)
return M{"result": "SUCCESS"}, nil
}
@ -85,7 +85,7 @@ func XaTransOut(c *gin.Context) (interface{}, error) {
Update("balance", gorm.Expr("balance + ?", req.Amount))
return dbr.Error
})
common.PanicIfError(err)
e2p(err)
return M{"result": "SUCCESS"}, nil
}

21
xa.go
View File

@ -10,6 +10,9 @@ import (
)
type M = map[string]interface{}
var e2p = common.E2P
type XaGlobalFunc func() error
type XaLocalFunc func(db *common.MyDb) error
@ -27,7 +30,7 @@ func XaClientNew(server string, mysqlConf map[string]string, app *gin.Engine, ca
CallbackUrl: callbackUrl,
}
u, err := url.Parse(callbackUrl)
common.PanicIfError(err)
e2p(err)
app.POST(u.Path, common.WrapHandler(func(c *gin.Context) (interface{}, error) {
type CallbackReq struct {
Gid string `json:"gid"`
@ -36,7 +39,7 @@ func XaClientNew(server string, mysqlConf map[string]string, app *gin.Engine, ca
}
req := CallbackReq{}
b, err := c.GetRawData()
common.PanicIfError(err)
e2p(err)
common.MustUnmarshal(b, &req)
tx, my := common.DbAlone(xa.Conf)
defer func() {
@ -54,19 +57,19 @@ func XaClientNew(server string, mysqlConf map[string]string, app *gin.Engine, ca
return xa
}
func (xa *XaClient) XaLocalTransaction(gid string, transFunc XaLocalFunc) (rerr error) {
defer common.Panic2Error(&rerr)
defer common.P2E(&rerr)
branch := common.GenGid()
tx, my := common.DbAlone(xa.Conf)
defer func() { my.Close() }()
tx.Must().Exec(fmt.Sprintf("XA start '%s'", branch))
err := transFunc(tx)
common.PanicIfError(err)
e2p(err)
resp, err := common.RestyClient.R().
SetBody(&M{"gid": gid, "branch": branch, "trans_type": "xa", "status": "prepared", "url": xa.CallbackUrl}).
Post(xa.Server + "/branch")
common.PanicIfError(err)
e2p(err)
if !strings.Contains(resp.String(), "SUCCESS") {
common.PanicIfError(fmt.Errorf("unknown server response: %s", resp.String()))
e2p(fmt.Errorf("unknown server response: %s", resp.String()))
}
tx.Must().Exec(fmt.Sprintf("XA end '%s'", branch))
tx.Must().Exec(fmt.Sprintf("XA prepare '%s'", branch))
@ -86,14 +89,14 @@ func (xa *XaClient) XaGlobalTransaction(gid string, transFunc XaGlobalFunc) (rer
}
}()
resp, err := common.RestyClient.R().SetBody(data).Post(xa.Server + "/prepare")
common.PanicIfError(err)
e2p(err)
if !strings.Contains(resp.String(), "SUCCESS") {
panic(fmt.Errorf("unexpected result: %s", resp.String()))
}
err = transFunc()
common.PanicIfError(err)
e2p(err)
resp, err = common.RestyClient.R().SetBody(data).Post(xa.Server + "/commit")
common.PanicIfError(err)
e2p(err)
if !strings.Contains(resp.String(), "SUCCESS") {
panic(fmt.Errorf("unexpected result: %s", resp.String()))
}