Feature/MSG delayed consumption MSG模式增加延时消费功能

client/dtmgrpc/dtmgpb/dtmgimp.proto

@@ -2,6 +2,7 @@ syntax = "proto3";
 
 option go_package = "./dtmgpb";
 import "google/protobuf/empty.proto";
+import "google/protobuf/timestamp.proto";
 
 package dtmgimp;
 
@@ -39,6 +40,7 @@ message DtmRequest {
   string Steps = 7;
   map<string, string> ReqExtra = 8;
   string RollbackReason = 9;
+  google.protobuf.Timestamp NextCronTime = 10;
 }
 
 message DtmGidReply {

client/dtmgrpc/dtmgpb/generat.sh

@@ -0,0 +1,30 @@
+# 步骤 1：先获取并配置 GOPATH（临时生效）
+# 获取 GOPATH 并赋值给变量（自动适配你的路径）
+GOPATH=$(go env GOPATH)
+# 将 GOPATH/bin 加入 PATH（解决插件找不到的问题）
+export PATH=$PATH:$GOPATH/bin
+
+# 步骤 2：验证插件是否存在（可选，但建议执行）
+# 检查 protoc-gen-go 是否存在
+if [ -f "$GOPATH/bin/protoc-gen-go" ]; then
+  echo "protoc-gen-go 插件存在"
+else
+  echo "protoc-gen-go 插件不存在，正在安装..."
+  go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
+fi
+
+# 检查 protoc-gen-go-grpc 是否存在
+if [ -f "$GOPATH/bin/protoc-gen-go-grpc" ]; then
+  echo "protoc-gen-go-grpc 插件存在"
+else
+  echo "protoc-gen-go-grpc 插件不存在，正在安装兼容版本..."
+  # 使用与项目gRPC v1.56.3兼容的版本
+  go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.3.0
+fi
+
+
+# 步骤 3：生成 Go 代码（核心命令）
+# 生成基础 pb.go 文件
+protoc --go_out=. --go_opt=paths=source_relative dtmgimp.proto
+# 生成 grpc 相关 pb.go 文件
+protoc --go-grpc_out=. --go-grpc_opt=paths=source_relative dtmgimp.proto

dtmsvr/api.go

@@ -39,6 +39,13 @@ func svcSubmit(t *TransGlobal) interface{} {
 	} else if err != nil {
 		return err
 	}
+
+	// In msg mode, allow delaying consumption by using a custom NextCronTime
+	// WaitResult=true and NextCronTime are mutually exclusive, WaitResult takes priority
+	if !t.WaitResult && t.TransType == "msg" && !t.NextCronTime.IsZero() && t.NextCronTime.After(time.Now().Add(3*time.Second)) {
+		return nil
+	}
+
 	return t.Process(branches)
 }
 

dtmsvr/trans_class.go

@@ -85,6 +85,7 @@ func TransFromDtmRequest(ctx context.Context, c *dtmgpb.DtmRequest) *TransGlobal
 	if c.TransOptions != nil {
 		o = c.TransOptions
 	}
+	nextCronTime := c.NextCronTime.AsTime()
 	r := TransGlobal{TransGlobalStore: storage.TransGlobalStore{
 		Gid:            c.Gid,
 		TransType:      c.TransType,
@@ -93,6 +94,7 @@ func TransFromDtmRequest(ctx context.Context, c *dtmgpb.DtmRequest) *TransGlobal
 		BinPayloads:    c.BinPayloads,
 		CustomData:     c.CustomedData,
 		RollbackReason: c.RollbackReason,
+		NextCronTime:   &nextCronTime,
 		TransOptions: dtmcli.TransOptions{
 			WaitResult:     o.WaitResult,
 			TimeoutToFail:  o.TimeoutToFail,

dtmsvr/trans_process.go

@@ -77,7 +77,9 @@ func (t *TransGlobal) processInner(ctx context.Context, branches []TransBranch)
 
 func (t *TransGlobal) saveNew() ([]TransBranch, error) {
 	t.NextCronInterval = t.getNextCronInterval(cronReset)
-	t.NextCronTime = dtmutil.GetNextTime(t.NextCronInterval)
+	if t.NextCronTime == nil || t.NextCronTime.IsZero() {
+		t.NextCronTime = dtmutil.GetNextTime(t.NextCronInterval)
+	}
 	t.ExtData = dtmimp.MustMarshalString(t.Ext)
 	if t.ExtData == "{}" {
 		t.ExtData = ""