Merge pull request #199 from NeptuneG/main

支持grpc下SAGA的并发执行

dtmcli/saga.go

@@ -43,8 +43,13 @@ func (s *Saga) EnableConcurrent() *Saga {
 
 // Submit submit the saga trans
 func (s *Saga) Submit() error {
+	s.AddConcurrentContext()
+	return dtmimp.TransCallDtm(&s.TransBase, s, "submit")
+}
+
+// AddConcurrentContext adds concurrent options to the request context
+func (s *Saga) AddConcurrentContext() {
 	if s.concurrent {
 		s.CustomData = dtmimp.MustMarshalString(map[string]interface{}{"orders": s.orders, "concurrent": s.concurrent})
 	}
-	return dtmimp.TransCallDtm(&s.TransBase, s, "submit")
 }

dtmgrpc/saga.go

@@ -43,5 +43,6 @@ func (s *SagaGrpc) EnableConcurrent() *SagaGrpc {
 
 // Submit submit the saga trans
 func (s *SagaGrpc) Submit() error {
+	s.Saga.AddConcurrentContext()
 	return dtmgimp.DtmGrpcCall(&s.Saga.TransBase, "Submit")
 }

dtmsvr/trans_class.go

@@ -96,6 +96,7 @@ func TransFromDtmRequest(ctx context.Context, c *dtmgpb.DtmRequest) *TransGlobal
 		QueryPrepared: c.QueryPrepared,
 		Protocol:      "grpc",
 		BinPayloads:   c.BinPayloads,
+		CustomData:    c.CustomedData,
 		TransOptions: dtmcli.TransOptions{
 			WaitResult:         o.WaitResult,
 			TimeoutToFail:      o.TimeoutToFail,