unit test seems ok

common/utils.go

@@ -1,9 +1,13 @@
 package common
 
 import (
+	"bytes"
 	"encoding/json"
+	"io/ioutil"
+	"time"
 
 	"github.com/bwmarrin/snowflake"
+	"github.com/gin-gonic/gin"
 	"github.com/go-resty/resty/v2"
 	"github.com/sirupsen/logrus"
 )
@@ -66,3 +70,26 @@ func init() {
 		return nil
 	})
 }
+
+func GetGinApp() *gin.Engine {
+	gin.SetMode(gin.ReleaseMode)
+	app := gin.Default()
+	app.Use(func(c *gin.Context) {
+		body := ""
+		if c.Request.Method == "POST" {
+			rb, err := c.GetRawData()
+			if err != nil {
+				logrus.Printf("GetRawData error: %s", err.Error())
+			} else {
+				body = string(rb)
+				c.Request.Body = ioutil.NopCloser(bytes.NewBuffer(rb))
+			}
+		}
+		began := time.Now()
+		logrus.Printf("begin %s %s query: %s body: %s", c.Request.Method, c.FullPath(), c.Request.URL.RawQuery, body)
+		c.Next()
+		logrus.Printf("used %d ms %s %s query: %s body: %s", time.Since(began).Milliseconds(), c.Request.Method, c.FullPath(), c.Request.URL.RawQuery, body)
+
+	})
+	return app
+}

dtm/saga.go

@@ -3,42 +3,42 @@ package dtm
 import (
 	"fmt"
 
-	"github.com/gin-gonic/gin"
 	"github.com/sirupsen/logrus"
 	"github.com/yedf/dtm/common"
 )
 
+type Saga struct {
+	SagaData
+	Server string
+}
+
 type SagaData struct {
 	Gid        string     `json:"gid"`
 	Steps      []SagaStep `json:"steps"`
 	TransQuery string     `json:"trans_query"`
 }
-type Saga struct {
-	SagaData
-	Server string
-}
 type SagaStep struct {
 	Action     string `json:"action"`
 	Compensate string `json:"compensate"`
-	PostData   gin.H  `json:"post_data"`
+	PostData   string `json:"post_data"`
 }
 
-func SagaNew(server string, gid string) *Saga {
+func SagaNew(server string, gid string, transQuery string) *Saga {
 	return &Saga{
 		SagaData: SagaData{
-			Gid: gid,
+			Gid:        gid,
+			TransQuery: transQuery,
 		},
 		Server: server,
 	}
 }
-func (s *Saga) Add(action string, compensate string, postData gin.H) error {
+func (s *Saga) Add(action string, compensate string, postData interface{}) error {
 	logrus.Printf("saga %s Add %s %s %v", s.Gid, action, compensate, postData)
 	step := SagaStep{
 		Action:     action,
 		Compensate: compensate,
-		PostData:   postData,
+		PostData:   common.MustMarshalString(postData),
 	}
-	step.PostData = postData
 	s.Steps = append(s.Steps, step)
 	return nil
 }
@@ -47,8 +47,7 @@ func (s *Saga) getBody() *SagaData {
 	return &s.SagaData
 }
 
-func (s *Saga) Prepare(url string) error {
+func (s *Saga) Prepare() error {
-	s.TransQuery = url
 	logrus.Printf("preparing %s body: %v", s.Gid, s.getBody())
 	resp, err := common.RestyClient.R().SetBody(s.getBody()).Post(fmt.Sprintf("%s/prepare", s.Server))
 	if err != nil {

dtm_test.go

@@ -2,10 +2,15 @@ package main
 
 import (
 	"testing"
+	"time"
 
 	"github.com/go-playground/assert/v2"
+	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
+	"github.com/yedf/dtm/common"
+	"github.com/yedf/dtm/dtm"
 	"github.com/yedf/dtm/dtmsvr"
+	"github.com/yedf/dtm/examples"
 )
 
 func init() {
@@ -16,8 +21,73 @@ func TestViper(t *testing.T) {
 	assert.Equal(t, "test_val", viper.GetString("test"))
 }
 
-func TTestDtmSvr(t *testing.T) {
+func TestDtmSvr(t *testing.T) {
+	// 清理数据
+	rabbit := dtmsvr.RabbitmqNew(&dtmsvr.ServerConfig.Rabbitmq)
+	queprepared := rabbit.QueueNew(dtmsvr.RabbitmqConstPrepared)
+	for i := 0; i < queprepared.Queue.Messages; i++ {
+		queprepared.WaitAndHandleOne(func(data M) {
+			logrus.Printf("ignoring prepared queue data before test")
+		})
+	}
+	quecommited := rabbit.QueueNew(dtmsvr.RabbitmqConstCommited)
+	for i := 0; i < quecommited.Queue.Messages; i++ {
+		quecommited.WaitAndHandleOne(func(data M) {
+			logrus.Printf("ignoring commited queue data before test")
+		})
+	}
+	db := dtmsvr.DbGet()
+	common.PanicIfError(db.Exec("truncate test1.a_saga").Error)
+	common.PanicIfError(db.Exec("truncate test1.a_saga_step").Error)
+
+	// 启动组件
+	go dtmsvr.StartSvr()
+	go examples.StartSvr()
+	time.Sleep(time.Duration(100 * 1000 * 1000))
+
+	// 开始第一个正常流程的测试
+	saga := genSaga("gid-1", false, false)
+	saga.Prepare()
+	queprepared.WaitAndHandleOne(dtmsvr.HandlePreparedMsg)
+	sm := dtmsvr.SagaModel{}
+	db.Model(&sm).Where("gid=?", saga.Gid).First(&sm)
+	assert.Equal(t, "prepared", sm.Status)
+	saga.Commit()
+	quecommited.WaitAndHandleOne(dtmsvr.HandleCommitedMsg)
+	db.Model(&dtmsvr.SagaModel{}).Where("gid=?", saga.Gid).First(&sm)
+	assert.Equal(t, "finished", sm.Status)
+	steps := []dtmsvr.SagaStepModel{}
+	db.Model(&dtmsvr.SagaStepModel{}).Where("gid=?", saga.Gid).Find(&steps)
+	assert.Equal(t, true, steps[0].Status == "pending" && steps[2].Status == "pending" && steps[1].Status == "finished" && steps[3].Status == "finished")
+
+	saga = genSaga("gid-2", false, true)
+	saga.Commit()
+	quecommited.WaitAndHandleOne(dtmsvr.HandleCommitedMsg)
+	saga.Prepare()
+	queprepared.WaitAndHandleOne(dtmsvr.HandlePreparedMsg)
+	sm = dtmsvr.SagaModel{}
+	db.Model(&dtmsvr.SagaModel{}).Where("gid=?", saga.Gid).First(&sm)
+	assert.Equal(t, "rollbacked", sm.Status)
+	steps = []dtmsvr.SagaStepModel{}
+	db.Model(&dtmsvr.SagaStepModel{}).Where("gid=?", saga.Gid).Find(&steps)
+	assert.Equal(t, true, steps[0].Status == "rollbacked" && steps[2].Status == "rollbacked" && steps[1].Status == "finished" && steps[3].Status == "rollbacked")
+
+	// assert.Equal(t, 1, 0)
+	// 开始测试
+
 	// 发送Prepare请求后，验证数据库
 	// ConsumeHalfMsg 验证数据库
 	// ConsumeMsg 验证数据库
 }
+
+func genSaga(gid string, inFailed bool, outFailed bool) *dtm.Saga {
+	saga := dtm.SagaNew(examples.TcServer, gid, examples.BusiApi+"/TransQuery")
+	req := examples.TransReq{
+		Amount:         30,
+		TransInFailed:  inFailed,
+		TransOutFailed: outFailed,
+	}
+	saga.Add(examples.Busi+"/TransIn", examples.Busi+"/TransInCompensate", &req)
+	saga.Add(examples.Busi+"/TransOut", examples.Busi+"/TransOutCompensate", &req)
+	return saga
+}

dtmsvr/consumer.go

@@ -5,7 +5,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/gin-gonic/gin"
 	"github.com/sirupsen/logrus"
 	"github.com/yedf/dtm/common"
 	"gorm.io/gorm"
@@ -47,7 +46,7 @@ func (*SagaStepModel) TableName() string {
 	return "test1.a_saga_step"
 }
 
-func handlePreparedMsg(data gin.H) {
+func HandlePreparedMsg(data M) {
 	db := DbGet()
 	logrus.Printf("creating saga model in prepare")
 	data["steps"] = common.MustMarshalString(data["steps"])
@@ -60,7 +59,7 @@ func handlePreparedMsg(data gin.H) {
 	}).Create(&m)
 }
 
-func handleCommitedMsg(data gin.H) {
+func HandleCommitedMsg(data M) {
 	db := DbGet()
 	logrus.Printf("creating saga model in commited")
 	steps := data["steps"].([]interface{})
@@ -83,7 +82,7 @@ func handleCommitedMsg(data gin.H) {
 			nsteps = append(nsteps, SagaStepModel{
 				Gid:    m.Gid,
 				Step:   len(nsteps) + 1,
-				Data:   common.MustMarshalString(step["post_data"]),
+				Data:   step["post_data"].(string),
 				Url:    step["compensate"].(string),
 				Type:   "compensate",
 				Status: "pending",
@@ -91,7 +90,7 @@ func handleCommitedMsg(data gin.H) {
 			nsteps = append(nsteps, SagaStepModel{
 				Gid:    m.Gid,
 				Step:   len(nsteps) + 1,
-				Data:   common.MustMarshalString(step["post_data"]),
+				Data:   step["post_data"].(string),
 				Url:    step["action"].(string),
 				Type:   "action",
 				Status: "pending",
@@ -169,8 +168,7 @@ func ProcessCommitedSaga(gid string) (rerr error) {
 				checkAndCommit(dbr)
 				break
 			} else {
-				logrus.Errorf("unknown response: %s, will be retried", body)
+				return fmt.Errorf("unknown response: %s, will be retried", body)
-				break
 			}
 		}
 	}
@@ -219,7 +217,7 @@ func StartConsumePreparedMsg(consumers int) {
 	for i := 0; i < consumers; i++ {
 		go func() {
 			que := r.QueueNew(RabbitmqConstPrepared)
-			que.WaitAndHandle(handlePreparedMsg)
+			que.WaitAndHandle(HandlePreparedMsg)
 		}()
 	}
 }
@@ -230,7 +228,7 @@ func StartConsumeCommitedMsg(consumers int) {
 	for i := 0; i < consumers; i++ {
 		go func() {
 			que := r.QueueNew(RabbitmqConstCommited)
-			que.WaitAndHandle(handleCommitedMsg)
+			que.WaitAndHandle(HandleCommitedMsg)
 		}()
 	}
 }

dtmsvr/rabbitmq.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"sync"
 
-	"github.com/gin-gonic/gin"
 	"github.com/sirupsen/logrus"
 	"github.com/streadway/amqp"
 	"github.com/yedf/dtm/common"
@@ -107,6 +106,7 @@ func (r *Rabbitmq) SendAndConfirm(key RabbitmqConst, data map[string]interface{}
 
 type RabbitmqQueue struct {
 	Name       string
+	Queue      *amqp.Queue
 	Channel    *amqp.Channel
 	Conn       *amqp.Connection
 	Deliveries <-chan amqp.Delivery
@@ -117,12 +117,12 @@ func (q *RabbitmqQueue) Close() {
 	// q.Conn.Close()
 }
 
-func (q *RabbitmqQueue) WaitAndHandle(handler func(data gin.H)) {
+func (q *RabbitmqQueue) WaitAndHandle(handler func(data M)) {
 	for {
 		q.WaitAndHandleOne(handler)
 	}
 }
-func (q *RabbitmqQueue) WaitAndHandleOne(handler func(data gin.H)) {
+func (q *RabbitmqQueue) WaitAndHandleOne(handler func(data M)) {
 	logrus.Printf("%s reading message", q.Name)
 	msg := <-q.Deliveries
 	data := map[string]interface{}{}
@@ -168,6 +168,7 @@ func (r *Rabbitmq) QueueNew(queueType RabbitmqConst) *RabbitmqQueue {
 	)
 	common.PanicIfError(err)
 	return &RabbitmqQueue{
+		Queue:      &queue,
 		Name:       queueName,
 		Channel:    channel,
 		Deliveries: deliveries,

dtmsvr/rabbitmq_test.go

@@ -3,7 +3,6 @@ package dtmsvr
 import (
 	"testing"
 
-	"github.com/gin-gonic/gin"
 	"github.com/magiconair/properties/assert"
 	"github.com/sirupsen/logrus"
 	"github.com/yedf/dtm/common"
@@ -19,12 +18,12 @@ func TestRabbitConfig(t *testing.T) {
 
 func TestRabbitmq1Msg(t *testing.T) {
 	rb := RabbitmqNew(&ServerConfig.Rabbitmq)
-	err := rb.SendAndConfirm(RabbitmqConstPrepared, gin.H{
+	err := rb.SendAndConfirm(RabbitmqConstPrepared, M{
 		"gid": common.GenGid(),
 	})
 	assert.Equal(t, nil, err)
 	queue := rb.QueueNew(RabbitmqConstPrepared)
-	queue.WaitAndHandle(func(data gin.H) {
+	queue.WaitAndHandle(func(data M) {
 		logrus.Printf("processed msg: %v in queue1", data)
 	})
 	assert.Equal(t, 0, 1)

dtmsvr/service.go

@@ -11,7 +11,7 @@ func AddRoute(engine *gin.Engine) {
 }
 
 func Prepare(c *gin.Context) {
-	data := gin.H{}
+	data := M{}
 	err := c.BindJSON(&data)
 	if err != nil {
 		return
@@ -19,11 +19,11 @@ func Prepare(c *gin.Context) {
 	rabbit := RabbitmqGet()
 	err = rabbit.SendAndConfirm(RabbitmqConstPrepared, data)
 	common.PanicIfError(err)
-	c.JSON(200, gin.H{"message": "SUCCESS"})
+	c.JSON(200, M{"message": "SUCCESS"})
 }
 
 func Commit(c *gin.Context) {
-	data := gin.H{}
+	data := M{}
 	err := c.BindJSON(&data)
 	if err != nil {
 		return
@@ -31,5 +31,5 @@ func Commit(c *gin.Context) {
 	rabbit := RabbitmqGet()
 	err = rabbit.SendAndConfirm(RabbitmqConstCommited, data)
 	common.PanicIfError(err)
-	c.JSON(200, gin.H{"message": "SUCCESS"})
+	c.JSON(200, M{"message": "SUCCESS"})
 }

dtmsvr/svr.go

@@ -1,38 +1,21 @@
 package dtmsvr
 
 import (
-	"bytes"
-	"io/ioutil"
-	"time"
-
-	"github.com/gin-gonic/gin"
 	"github.com/sirupsen/logrus"
+	"github.com/yedf/dtm/common"
 )
 
 func Main() {
-	logrus.Printf("start dtmsvr")
+	StartConsumePreparedMsg(1)
-	gin.SetMode(gin.ReleaseMode)
-	app := gin.Default()
-	app.Use(func(c *gin.Context) {
-		body := ""
-		if c.Request.Method == "POST" {
-			rb, err := c.GetRawData()
-			if err != nil {
-				logrus.Printf("GetRawData error: %s", err.Error())
-			} else {
-				body = string(rb)
-				c.Request.Body = ioutil.NopCloser(bytes.NewBuffer(rb))
-			}
-		}
-		began := time.Now()
-		logrus.Printf("begin %s %s query: %s body: %s", c.Request.Method, c.FullPath(), c.Request.URL.RawQuery, body)
-		c.Next()
-		logrus.Printf("used %d ms %s %s query: %s", time.Since(began).Milliseconds(), c.Request.Method, c.FullPath(), c.Request.URL.RawQuery)
-
-	})
-	AddRoute(app)
-	// StartConsumePreparedMsg(1)
 	StartConsumeCommitedMsg(1)
 	logrus.Printf("dtmsvr listen at: 8080")
-	go app.Run()
+	go StartSvr()
+}
+
+func StartSvr() {
+	logrus.Printf("start dtmsvr")
+	app := common.GetGinApp()
+	AddRoute(app)
+	logrus.Printf("dtmsvr listen at: 8080")
+	app.Run()
 }

examples/cli.go

@@ -1,25 +1,27 @@
 package examples
 
 import (
-	"github.com/gin-gonic/gin"
 	"github.com/sirupsen/logrus"
+	"github.com/yedf/dtm/common"
 )
 
 func Main() {
-	logrus.Printf("examples starting")
+	go StartSvr()
-	gin.SetMode(gin.ReleaseMode)
-	app := gin.Default()
-	app.POST(BusiApi+"/TransIn", TransIn)
-	app.POST(BusiApi+"/TransInCompensate", TransInCompensate)
-	app.POST(BusiApi+"/TransOut", TransOut)
-	app.POST(BusiApi+"/TransOutCompensate", TransOutCompensate)
-	app.POST(BusiApi+"/TransQuery", TransQuery)
-
-	go app.Run(":8081")
-	logrus.Printf("examples istening at %d", BusiPort)
 	trans(&TransReq{
 		Amount:         30,
 		TransInFailed:  false,
 		TransOutFailed: true,
 	})
 }
+
+func StartSvr() {
+	logrus.Printf("examples starting")
+	app := common.GetGinApp()
+	app.POST(BusiApi+"/TransIn", TransIn)
+	app.POST(BusiApi+"/TransInCompensate", TransInCompensate)
+	app.POST(BusiApi+"/TransOut", TransOut)
+	app.POST(BusiApi+"/TransOutCompensate", TransOutCompensate)
+	app.POST(BusiApi+"/TransQuery", TransQuery)
+	logrus.Printf("examples istening at %d", BusiPort)
+	app.Run(":8081")
+}

examples/saga.go

@@ -4,11 +4,11 @@ import (
 	"fmt"
 
 	"github.com/gin-gonic/gin"
-	"github.com/gin-gonic/gin/binding"
 	"github.com/sirupsen/logrus"
 	"github.com/yedf/dtm/dtm"
 )
 
+type M = map[string]interface{}
 type TransReq struct {
 	Amount         int  `json:"amount"`
 	TransInFailed  bool `json:"transInFailed"`
@@ -27,7 +27,7 @@ func TransIn(c *gin.Context) {
 		c.Error(fmt.Errorf("TransIn failed for gid: %s", gid))
 		return
 	}
-	c.JSON(200, gin.H{"result": "SUCCESS"})
+	c.JSON(200, M{"result": "SUCCESS"})
 }
 
 func TransInCompensate(c *gin.Context) {
@@ -37,25 +37,22 @@ func TransInCompensate(c *gin.Context) {
 		return
 	}
 	logrus.Printf("%s TransInCompensate: %v", gid, req)
-	c.JSON(200, gin.H{"result": "SUCCESS"})
+	c.JSON(200, M{"result": "SUCCESS"})
 }
 
 func TransOut(c *gin.Context) {
 	gid := c.Query("gid")
 	req := TransReq{}
-	req2 := TransReq{}
-	c.ShouldBindBodyWith(&req2, binding.JSON)
-	c.ShouldBindBodyWith(&req, binding.JSON)
 	if err := c.BindJSON(&req); err != nil {
 		return
 	}
 	logrus.Printf("%s TransOut: %v", gid, req)
 	if req.TransOutFailed {
 		logrus.Printf("%s TransOut %v failed", gid, req)
-		c.JSON(500, gin.H{"result": "FAIL"})
+		c.JSON(500, M{"result": "FAIL"})
 		return
 	}
-	c.JSON(200, gin.H{"result": "SUCCESS"})
+	c.JSON(200, M{"result": "SUCCESS"})
 }
 
 func TransOutCompensate(c *gin.Context) {
@@ -65,7 +62,7 @@ func TransOutCompensate(c *gin.Context) {
 		return
 	}
 	logrus.Printf("%s TransOutCompensate: %v", gid, req)
-	c.JSON(200, gin.H{"result": "SUCCESS"})
+	c.JSON(200, M{"result": "SUCCESS"})
 }
 
 func TransQuery(c *gin.Context) {
@@ -75,26 +72,26 @@ func TransQuery(c *gin.Context) {
 		return
 	}
 	logrus.Printf("%s TransQuery: %v", gid, req)
-	c.JSON(200, gin.H{"result": "SUCCESS"})
+	c.JSON(200, M{"result": "SUCCESS"})
 }
 
 func trans(req *TransReq) {
 	// gid := common.GenGid()
 	gid := "4eHhkCxVsQ1"
 	logrus.Printf("busi transaction begin: %s", gid)
-	saga := dtm.SagaNew(TcServer, gid)
+	saga := dtm.SagaNew(TcServer, gid, Busi+"/TransQuery")
 
-	saga.Add(Busi+"/TransIn", Busi+"/TransInCompensate", gin.H{
+	saga.Add(Busi+"/TransIn", Busi+"/TransInCompensate", M{
 		"amount":         req.Amount,
 		"transInFailed":  req.TransInFailed,
 		"transOutFailed": req.TransOutFailed,
 	})
-	saga.Add(Busi+"/TransOut", Busi+"/TransOutCompensate", gin.H{
+	saga.Add(Busi+"/TransOut", Busi+"/TransOutCompensate", M{
 		"amount":         req.Amount,
 		"transInFailed":  req.TransInFailed,
 		"transOutFailed": req.TransOutFailed,
 	})
-	saga.Prepare(Busi + "/TransQuery")
+	saga.Prepare()
 	logrus.Printf("busi trans commit")
 	saga.Commit()
 }

main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"encoding/json"
 	"time"
 
 	"github.com/yedf/dtm/common"
@@ -8,8 +9,17 @@ import (
 	"github.com/yedf/dtm/examples"
 )
 
+type M = map[string]interface{}
+
 func main() {
 	dtmsvr.LoadConfig()
+
+	s := common.MustMarshalString(M{
+		"a": 1,
+		"b": "str",
+	})
+	var obj interface{}
+	json.Unmarshal([]byte(s), &obj)
 	db := dtmsvr.DbGet()
 	tx := db.Begin()
 	common.PanicIfError(tx.Error)
@@ -25,7 +35,7 @@ func main() {
 	// logrus.SetFormatter(&logrus.JSONFormatter{})
 	// dtmsvr.LoadConfig()
 	// rb := dtmsvr.RabbitmqNew(&dtmsvr.ServerConfig.Rabbitmq)
-	// err := rb.SendAndConfirm(dtmsvr.RabbitmqConstPrepared, gin.H{
+	// err := rb.SendAndConfirm(dtmsvr.RabbitmqConstPrepared, M{
 	// 	"gid": common.GenGid(),
 	// })
 	// common.PanicIfError(err)