diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 17c6e77..d9fb0d6 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -1,6 +1,6 @@ name: Release on: - create: + push: tags: - 'v*.*.*' diff --git a/conf.sample.yml b/conf.sample.yml index a4e869c..673c9e9 100644 --- a/conf.sample.yml +++ b/conf.sample.yml @@ -56,6 +56,7 @@ # HttpPort: 36789 # GrpcPort: 36790 +# JsonRpcPort: 36791 ### advanced options # UpdateBranchAsyncGoroutineNum: 1 # num of async goroutine to update branch status diff --git a/dtmcli/dtmimp/trans_base.go b/dtmcli/dtmimp/trans_base.go index 879ffc2..4ec247b 100644 --- a/dtmcli/dtmimp/trans_base.go +++ b/dtmcli/dtmimp/trans_base.go @@ -65,6 +65,7 @@ type TransBase struct { Op string `json:"-"` // used in XA/TCC QueryPrepared string `json:"query_prepared,omitempty"` // used in MSG + Protocol string `json:"protocol"` } // NewTransBase new a TransBase @@ -93,6 +94,25 @@ func TransCallDtm(tb *TransBase, body interface{}, operation string) error { if tb.RequestTimeout != 0 { RestyClient.SetTimeout(time.Duration(tb.RequestTimeout) * time.Second) } + if tb.Protocol == "json-rpc" { + var result map[string]interface{} + resp, err := RestyClient.R(). + SetBody(map[string]interface{}{ + "jsonrpc": "2.0", + "id": "no-use", + "method": operation, + "params": body, + }). + SetResult(&result). + Post(tb.Dtm) + if err != nil { + return err + } + if resp.StatusCode() != http.StatusOK || result["error"] != nil { + return errors.New(resp.String()) + } + return nil + } resp, err := RestyClient.R(). SetBody(body).Post(fmt.Sprintf("%s/%s", tb.Dtm, operation)) if err != nil { @@ -118,6 +138,9 @@ func TransRegisterBranch(tb *TransBase, added map[string]string, operation strin // TransRequestBranch TransBase request branch result func TransRequestBranch(t *TransBase, method string, body interface{}, branchID string, op string, url string) (*resty.Response, error) { + if url == "" { + return nil, nil + } resp, err := RestyClient.R(). SetBody(body). SetQueryParams(map[string]string{ diff --git a/dtmgrpc/dtmgimp/utils.go b/dtmgrpc/dtmgimp/utils.go index fb3eadf..17aaa11 100644 --- a/dtmgrpc/dtmgimp/utils.go +++ b/dtmgrpc/dtmgimp/utils.go @@ -36,6 +36,7 @@ func DtmGrpcCall(s *dtmimp.TransBase, operation string) error { RetryInterval: s.RetryInterval, PassthroughHeaders: s.PassthroughHeaders, BranchHeaders: s.BranchHeaders, + RequestTimeout: s.RequestTimeout, }, QueryPrepared: s.QueryPrepared, CustomedData: s.CustomData, @@ -100,3 +101,19 @@ func GetMetaFromContext(ctx context.Context, name string) string { md, _ := metadata.FromIncomingContext(ctx) return mdGet(md, name) } + +type requestTimeoutKey struct{} + +// RequestTimeoutFromContext returns requestTime of transOption option +func RequestTimeoutFromContext(ctx context.Context) int64 { + if v, ok := ctx.Value(requestTimeoutKey{}).(int64); ok { + return v + } + + return 0 +} + +// RequestTimeoutNewContext sets requestTimeout of transOption option to context +func RequestTimeoutNewContext(ctx context.Context, requestTimeout int64) context.Context { + return context.WithValue(ctx, requestTimeoutKey{}, requestTimeout) +} diff --git a/dtmgrpc/dtmgpb/dtmgimp.pb.go b/dtmgrpc/dtmgpb/dtmgimp.pb.go index 9247260..671c1c6 100644 --- a/dtmgrpc/dtmgpb/dtmgimp.pb.go +++ b/dtmgrpc/dtmgpb/dtmgimp.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.27.1 -// protoc v3.17.3 +// protoc v3.19.4 // source: dtmgrpc/dtmgpb/dtmgimp.proto package dtmgpb @@ -31,6 +31,7 @@ type DtmTransOptions struct { RetryInterval int64 `protobuf:"varint,3,opt,name=RetryInterval,proto3" json:"RetryInterval,omitempty"` PassthroughHeaders []string `protobuf:"bytes,4,rep,name=PassthroughHeaders,proto3" json:"PassthroughHeaders,omitempty"` BranchHeaders map[string]string `protobuf:"bytes,5,rep,name=BranchHeaders,proto3" json:"BranchHeaders,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` + RequestTimeout int64 `protobuf:"varint,6,opt,name=RequestTimeout,proto3" json:"RequestTimeout,omitempty"` } func (x *DtmTransOptions) Reset() { @@ -100,6 +101,13 @@ func (x *DtmTransOptions) GetBranchHeaders() map[string]string { return nil } +func (x *DtmTransOptions) GetRequestTimeout() int64 { + if x != nil { + return x.RequestTimeout + } + return 0 +} + // DtmRequest request sent to dtm server type DtmRequest struct { state protoimpl.MessageState @@ -337,7 +345,7 @@ var file_dtmgrpc_dtmgpb_dtmgimp_proto_rawDesc = []byte{ 0x2f, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x07, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x1a, 0x1b, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x65, 0x6d, 0x70, 0x74, 0x79, 0x2e, 0x70, - 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xc2, 0x02, 0x0a, 0x0f, 0x44, 0x74, 0x6d, 0x54, 0x72, 0x61, 0x6e, + 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xea, 0x02, 0x0a, 0x0f, 0x44, 0x74, 0x6d, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x1e, 0x0a, 0x0a, 0x57, 0x61, 0x69, 0x74, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0a, 0x57, 0x61, 0x69, 0x74, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x24, 0x0a, 0x0d, 0x54, 0x69, 0x6d, 0x65, @@ -353,66 +361,68 @@ var file_dtmgrpc_dtmgpb_dtmgimp_proto_rawDesc = []byte{ 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, 0x74, 0x6d, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x2e, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x0d, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, - 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x1a, 0x40, 0x0a, 0x12, 0x42, 0x72, 0x61, 0x6e, 0x63, - 0x68, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, - 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, - 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, - 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xfc, 0x01, 0x0a, 0x0a, 0x44, 0x74, - 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x47, 0x69, 0x64, 0x18, - 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x47, 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x54, 0x72, - 0x61, 0x6e, 0x73, 0x54, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x54, - 0x72, 0x61, 0x6e, 0x73, 0x54, 0x79, 0x70, 0x65, 0x12, 0x3c, 0x0a, 0x0c, 0x54, 0x72, 0x61, 0x6e, - 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x18, - 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, 0x74, 0x6d, 0x54, 0x72, 0x61, 0x6e, - 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x0c, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x4f, - 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x22, 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, - 0x65, 0x64, 0x44, 0x61, 0x74, 0x61, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x43, 0x75, - 0x73, 0x74, 0x6f, 0x6d, 0x65, 0x64, 0x44, 0x61, 0x74, 0x61, 0x12, 0x20, 0x0a, 0x0b, 0x42, 0x69, - 0x6e, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x73, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0c, 0x52, - 0x0b, 0x42, 0x69, 0x6e, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x73, 0x12, 0x24, 0x0a, 0x0d, - 0x51, 0x75, 0x65, 0x72, 0x79, 0x50, 0x72, 0x65, 0x70, 0x61, 0x72, 0x65, 0x64, 0x18, 0x06, 0x20, - 0x01, 0x28, 0x09, 0x52, 0x0d, 0x51, 0x75, 0x65, 0x72, 0x79, 0x50, 0x72, 0x65, 0x70, 0x61, 0x72, - 0x65, 0x64, 0x12, 0x14, 0x0a, 0x05, 0x53, 0x74, 0x65, 0x70, 0x73, 0x18, 0x07, 0x20, 0x01, 0x28, - 0x09, 0x52, 0x05, 0x53, 0x74, 0x65, 0x70, 0x73, 0x22, 0x1f, 0x0a, 0x0b, 0x44, 0x74, 0x6d, 0x47, - 0x69, 0x64, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x47, 0x69, 0x64, 0x18, 0x01, - 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x47, 0x69, 0x64, 0x22, 0x82, 0x02, 0x0a, 0x10, 0x44, 0x74, - 0x6d, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x10, - 0x0a, 0x03, 0x47, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x47, 0x69, 0x64, - 0x12, 0x1c, 0x0a, 0x09, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x54, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, - 0x01, 0x28, 0x09, 0x52, 0x09, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x54, 0x79, 0x70, 0x65, 0x12, 0x1a, - 0x0a, 0x08, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x49, 0x44, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, - 0x52, 0x08, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x49, 0x44, 0x12, 0x0e, 0x0a, 0x02, 0x4f, 0x70, - 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x4f, 0x70, 0x12, 0x37, 0x0a, 0x04, 0x44, 0x61, - 0x74, 0x61, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, - 0x6d, 0x70, 0x2e, 0x44, 0x74, 0x6d, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x52, 0x65, 0x71, 0x75, - 0x65, 0x73, 0x74, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x04, 0x44, - 0x61, 0x74, 0x61, 0x12, 0x20, 0x0a, 0x0b, 0x42, 0x75, 0x73, 0x69, 0x50, 0x61, 0x79, 0x6c, 0x6f, - 0x61, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0b, 0x42, 0x75, 0x73, 0x69, 0x50, 0x61, - 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x1a, 0x37, 0x0a, 0x09, 0x44, 0x61, 0x74, 0x61, 0x45, 0x6e, 0x74, - 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, - 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, - 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x32, 0xb1, - 0x02, 0x0a, 0x03, 0x44, 0x74, 0x6d, 0x12, 0x38, 0x0a, 0x06, 0x4e, 0x65, 0x77, 0x47, 0x69, 0x64, - 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, - 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x14, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, - 0x6d, 0x70, 0x2e, 0x44, 0x74, 0x6d, 0x47, 0x69, 0x64, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x22, 0x00, - 0x12, 0x37, 0x0a, 0x06, 0x53, 0x75, 0x62, 0x6d, 0x69, 0x74, 0x12, 0x13, 0x2e, 0x64, 0x74, 0x6d, - 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, - 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, - 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x38, 0x0a, 0x07, 0x50, 0x72, 0x65, - 0x70, 0x61, 0x72, 0x65, 0x12, 0x13, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, - 0x74, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, - 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, - 0x79, 0x22, 0x00, 0x12, 0x36, 0x0a, 0x05, 0x41, 0x62, 0x6f, 0x72, 0x74, 0x12, 0x13, 0x2e, 0x64, + 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x12, 0x26, 0x0a, 0x0e, 0x52, 0x65, 0x71, 0x75, 0x65, + 0x73, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x18, 0x06, 0x20, 0x01, 0x28, 0x03, 0x52, + 0x0e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x1a, + 0x40, 0x0a, 0x12, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, + 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, + 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, + 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, + 0x01, 0x22, 0xfc, 0x01, 0x0a, 0x0a, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, + 0x12, 0x10, 0x0a, 0x03, 0x47, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x47, + 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x54, 0x79, 0x70, 0x65, 0x18, + 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x54, 0x79, 0x70, 0x65, + 0x12, 0x3c, 0x0a, 0x0c, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, + 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x18, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, + 0x2e, 0x44, 0x74, 0x6d, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, + 0x52, 0x0c, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x12, 0x22, + 0x0a, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x65, 0x64, 0x44, 0x61, 0x74, 0x61, 0x18, 0x04, + 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x43, 0x75, 0x73, 0x74, 0x6f, 0x6d, 0x65, 0x64, 0x44, 0x61, + 0x74, 0x61, 0x12, 0x20, 0x0a, 0x0b, 0x42, 0x69, 0x6e, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, + 0x73, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0c, 0x52, 0x0b, 0x42, 0x69, 0x6e, 0x50, 0x61, 0x79, 0x6c, + 0x6f, 0x61, 0x64, 0x73, 0x12, 0x24, 0x0a, 0x0d, 0x51, 0x75, 0x65, 0x72, 0x79, 0x50, 0x72, 0x65, + 0x70, 0x61, 0x72, 0x65, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x51, 0x75, 0x65, + 0x72, 0x79, 0x50, 0x72, 0x65, 0x70, 0x61, 0x72, 0x65, 0x64, 0x12, 0x14, 0x0a, 0x05, 0x53, 0x74, + 0x65, 0x70, 0x73, 0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x53, 0x74, 0x65, 0x70, 0x73, + 0x22, 0x1f, 0x0a, 0x0b, 0x44, 0x74, 0x6d, 0x47, 0x69, 0x64, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, + 0x10, 0x0a, 0x03, 0x47, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x47, 0x69, + 0x64, 0x22, 0x82, 0x02, 0x0a, 0x10, 0x44, 0x74, 0x6d, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x52, + 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x47, 0x69, 0x64, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x03, 0x47, 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x54, 0x72, 0x61, 0x6e, + 0x73, 0x54, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x54, 0x72, 0x61, + 0x6e, 0x73, 0x54, 0x79, 0x70, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, + 0x49, 0x44, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, + 0x49, 0x44, 0x12, 0x0e, 0x0a, 0x02, 0x4f, 0x70, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, + 0x4f, 0x70, 0x12, 0x37, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, + 0x32, 0x23, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, 0x74, 0x6d, 0x42, 0x72, + 0x61, 0x6e, 0x63, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x2e, 0x44, 0x61, 0x74, 0x61, + 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x04, 0x44, 0x61, 0x74, 0x61, 0x12, 0x20, 0x0a, 0x0b, 0x42, + 0x75, 0x73, 0x69, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0c, + 0x52, 0x0b, 0x42, 0x75, 0x73, 0x69, 0x50, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x1a, 0x37, 0x0a, + 0x09, 0x44, 0x61, 0x74, 0x61, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, + 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, + 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, + 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x32, 0xb1, 0x02, 0x0a, 0x03, 0x44, 0x74, 0x6d, 0x12, 0x38, + 0x0a, 0x06, 0x4e, 0x65, 0x77, 0x47, 0x69, 0x64, 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, + 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, + 0x1a, 0x14, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, 0x74, 0x6d, 0x47, 0x69, + 0x64, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x22, 0x00, 0x12, 0x37, 0x0a, 0x06, 0x53, 0x75, 0x62, 0x6d, + 0x69, 0x74, 0x12, 0x13, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, 0x74, 0x6d, + 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, + 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, + 0x00, 0x12, 0x38, 0x0a, 0x07, 0x50, 0x72, 0x65, 0x70, 0x61, 0x72, 0x65, 0x12, 0x13, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, - 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x45, 0x0a, 0x0e, 0x52, - 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x12, 0x19, 0x2e, - 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, 0x74, 0x6d, 0x42, 0x72, 0x61, 0x6e, 0x63, - 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, - 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, - 0x22, 0x00, 0x42, 0x0a, 0x5a, 0x08, 0x2e, 0x2f, 0x64, 0x74, 0x6d, 0x67, 0x70, 0x62, 0x62, 0x06, - 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x12, 0x36, 0x0a, 0x05, 0x41, + 0x62, 0x6f, 0x72, 0x74, 0x12, 0x13, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, 0x44, + 0x74, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, + 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, + 0x79, 0x22, 0x00, 0x12, 0x45, 0x0a, 0x0e, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, 0x42, + 0x72, 0x61, 0x6e, 0x63, 0x68, 0x12, 0x19, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x69, 0x6d, 0x70, 0x2e, + 0x44, 0x74, 0x6d, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, + 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, + 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x42, 0x0a, 0x5a, 0x08, 0x2e, 0x2f, + 0x64, 0x74, 0x6d, 0x67, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/dtmgrpc/dtmgpb/dtmgimp.proto b/dtmgrpc/dtmgpb/dtmgimp.proto index f97b277..0a76d53 100644 --- a/dtmgrpc/dtmgpb/dtmgimp.proto +++ b/dtmgrpc/dtmgpb/dtmgimp.proto @@ -20,6 +20,7 @@ message DtmTransOptions { int64 RetryInterval = 3; repeated string PassthroughHeaders = 4; map BranchHeaders = 5; + int64 RequestTimeout = 6; } // DtmRequest request sent to dtm server diff --git a/dtmgrpc/dtmgpb/dtmgimp_grpc.pb.go b/dtmgrpc/dtmgpb/dtmgimp_grpc.pb.go index 8381443..db22f19 100644 --- a/dtmgrpc/dtmgpb/dtmgimp_grpc.pb.go +++ b/dtmgrpc/dtmgpb/dtmgimp_grpc.pb.go @@ -1,4 +1,8 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.2.0 +// - protoc v3.19.4 +// source: dtmgrpc/dtmgpb/dtmgimp.proto package dtmgpb diff --git a/dtmsvr/api_json_rpc.go b/dtmsvr/api_json_rpc.go new file mode 100644 index 0000000..d9f3936 --- /dev/null +++ b/dtmsvr/api_json_rpc.go @@ -0,0 +1,138 @@ +package dtmsvr + +import ( + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/dtm-labs/dtm/dtmcli" + "github.com/dtm-labs/dtm/dtmcli/dtmimp" + "github.com/dtm-labs/dtm/dtmcli/logger" + "github.com/gin-gonic/gin" +) + +const jrpcCodeFailure = -32901 +const jrpcCodeOngoing = -32902 + +type jrpcReq struct { + Method string `json:"method"` + Jsonrpc string `json:"jsonrpc"` + Params interface{} `json:"params"` + ID string `json:"id"` +} + +func addJrpcRouter(engine *gin.Engine) { + type jrpcFunc = func(interface{}) interface{} + handlers := map[string]jrpcFunc{ + "newGid": jrpcNewGid, + "prepare": jrpcPrepare, + "submit": jrpcSubmit, + "abort": jrpcAbort, + "registerBranch": jrpcRegisterBranch, + } + engine.POST("/api/json-rpc", func(c *gin.Context) { + began := time.Now() + var err error + var req jrpcReq + var jerr map[string]interface{} + r := func() interface{} { + defer dtmimp.P2E(&err) + err2 := c.BindJSON(&req) + if err2 != nil { + jerr = map[string]interface{}{ + "code": -32700, + "message": fmt.Sprintf("Parse json error: %s", err2.Error()), + } + } else if req.ID == "" || req.Jsonrpc != "2.0" { + jerr = map[string]interface{}{ + "code": -32600, + "message": fmt.Sprintf("Bad json request: %s", dtmimp.MustMarshalString(req)), + } + } else if handlers[req.Method] == nil { + jerr = map[string]interface{}{ + "code": -32601, + "message": fmt.Sprintf("Method not found: %s", req.Method), + } + } else if handlers[req.Method] != nil { + return handlers[req.Method](req.Params) + } + return nil + }() + + // error maybe returned in r, assign it to err + if ne, ok := r.(error); ok && err == nil { + err = ne + } + + if err != nil { + if errors.Is(err, dtmcli.ErrFailure) { + jerr = map[string]interface{}{ + "code": jrpcCodeFailure, + "message": err.Error(), + } + } else if errors.Is(err, dtmcli.ErrOngoing) { + jerr = map[string]interface{}{ + "code": jrpcCodeOngoing, + "message": err.Error(), + } + } else if jerr == nil { + jerr = map[string]interface{}{ + "code": -32603, + "message": err.Error(), + } + } + } + + result := map[string]interface{}{ + "jsonrpc": "2.0", + "id": req.ID, + "error": jerr, + "result": r, + } + b, _ := json.Marshal(result) + cont := string(b) + if jerr == nil || jerr["code"] == jrpcCodeOngoing { + logger.Infof("%2dms %d %s %s %s", time.Since(began).Milliseconds(), 200, c.Request.Method, c.Request.RequestURI, cont) + } else { + logger.Errorf("%2dms %d %s %s %s", time.Since(began).Milliseconds(), 200, c.Request.Method, c.Request.RequestURI, cont) + } + c.JSON(200, result) + }) +} + +// TransFromJrpcParams construct TransGlobal from jrpc params +func TransFromJrpcParams(params interface{}) *TransGlobal { + t := TransGlobal{} + dtmimp.MustRemarshal(params, &t) + t.setupPayloads() + return &t +} + +func jrpcNewGid(interface{}) interface{} { + return map[string]interface{}{"gid": GenGid()} +} + +func jrpcPrepare(params interface{}) interface{} { + return svcPrepare(TransFromJrpcParams(params)) +} + +func jrpcSubmit(params interface{}) interface{} { + return svcSubmit(TransFromJrpcParams(params)) +} + +func jrpcAbort(params interface{}) interface{} { + return svcAbort(TransFromJrpcParams(params)) +} + +func jrpcRegisterBranch(params interface{}) interface{} { + data := map[string]string{} + dtmimp.MustRemarshal(params, &data) + branch := TransBranch{ + Gid: data["gid"], + BranchID: data["branch_id"], + Status: dtmcli.StatusPrepared, + BinData: []byte(data["data"]), + } + return svcRegisterBranch(data["trans_type"], &branch, data) +} diff --git a/dtmsvr/config/config.go b/dtmsvr/config/config.go index 06a8fda..750af5e 100644 --- a/dtmsvr/config/config.go +++ b/dtmsvr/config/config.go @@ -76,6 +76,7 @@ type configType struct { RequestTimeout int64 `yaml:"RequestTimeout" default:"3"` HTTPPort int64 `yaml:"HttpPort" default:"36789"` GrpcPort int64 `yaml:"GrpcPort" default:"36790"` + JSONRPCPort int64 `yaml:"JsonRpcPort" default:"36791"` MicroService MicroService `yaml:"MicroService"` UpdateBranchSync int64 `yaml:"UpdateBranchSync"` UpdateBranchAsyncGoroutineNum int64 `yaml:"UpdateBranchAsyncGoroutineNum" default:"1"` diff --git a/dtmsvr/storage/registry/registry.go b/dtmsvr/storage/registry/registry.go index 003fd69..e699bd3 100644 --- a/dtmsvr/storage/registry/registry.go +++ b/dtmsvr/storage/registry/registry.go @@ -18,6 +18,12 @@ type StorageFactory interface { GetStorage() storage.Store } +var sqlFac = &SingletonFactory{ + creatorFunction: func() storage.Store { + return &sql.Store{} + }, +} + var storeFactorys = map[string]StorageFactory{ "boltdb": &SingletonFactory{ creatorFunction: func() storage.Store { @@ -29,16 +35,8 @@ var storeFactorys = map[string]StorageFactory{ return &redis.Store{} }, }, - "mysql": &SingletonFactory{ - creatorFunction: func() storage.Store { - return &sql.Store{} - }, - }, - "postgres": &SingletonFactory{ - creatorFunction: func() storage.Store { - return &sql.Store{} - }, - }, + "mysql": sqlFac, + "postgres": sqlFac, } // GetStore returns storage.Store diff --git a/dtmsvr/storage/sql/sql.go b/dtmsvr/storage/sql/sql.go index 566d2f8..d87297a 100644 --- a/dtmsvr/storage/sql/sql.go +++ b/dtmsvr/storage/sql/sql.go @@ -1,3 +1,9 @@ +/* + * Copyright (c) 2021 yedf. All rights reserved. + * Use of this source code is governed by a BSD-style + * license that can be found in the LICENSE file. + */ + package sql import ( diff --git a/dtmsvr/svr.go b/dtmsvr/svr.go index 1816087..4fafd89 100644 --- a/dtmsvr/svr.go +++ b/dtmsvr/svr.go @@ -12,9 +12,10 @@ import ( "net" "time" + "github.com/dtm-labs/dtm/dtmgrpc" + "github.com/dtm-labs/dtm/dtmcli" "github.com/dtm-labs/dtm/dtmcli/logger" - "github.com/dtm-labs/dtm/dtmgrpc" "github.com/dtm-labs/dtm/dtmgrpc/dtmgimp" "github.com/dtm-labs/dtm/dtmgrpc/dtmgpb" "github.com/dtm-labs/dtm/dtmutil" @@ -29,7 +30,11 @@ func StartSvr() { dtmcli.GetRestyClient().SetTimeout(time.Duration(conf.RequestTimeout) * time.Second) dtmgrpc.AddUnaryInterceptor(func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { - ctx2, cancel := context.WithTimeout(ctx, time.Duration(conf.RequestTimeout)*time.Second) + timeout := conf.RequestTimeout + if v := dtmgimp.RequestTimeoutFromContext(ctx); v != 0 { + timeout = v + } + ctx2, cancel := context.WithTimeout(ctx, time.Duration(timeout)*time.Second) defer cancel() return invoker(ctx2, method, req, reply, cc, opts...) }) @@ -38,7 +43,8 @@ func StartSvr() { app := dtmutil.GetGinApp() app = httpMetrics(app) addRoute(app) - logger.Infof("dtmsvr listen at: %d", conf.HTTPPort) + addJrpcRouter(app) + logger.Infof("dtmsvr http listen at: %d", conf.HTTPPort) go func() { err := app.Run(fmt.Sprintf(":%d", conf.HTTPPort)) if err != nil { diff --git a/dtmsvr/trans_class.go b/dtmsvr/trans_class.go index b6e080b..2eae58c 100644 --- a/dtmsvr/trans_class.go +++ b/dtmsvr/trans_class.go @@ -26,6 +26,22 @@ type TransGlobal struct { updateBranchSync bool } +func (t *TransGlobal) setupPayloads() { + // Payloads will be store in BinPayloads, Payloads is only used to Unmarshal + for _, p := range t.Payloads { + t.BinPayloads = append(t.BinPayloads, []byte(p)) + } + for _, d := range t.Steps { + if d["data"] != "" { + t.BinPayloads = append(t.BinPayloads, []byte(d["data"])) + } + } + if t.Protocol == "" { + t.Protocol = "http" + } + +} + // TransBranch branch transaction type TransBranch = storage.TransBranchStore @@ -61,17 +77,7 @@ func TransFromContext(c *gin.Context) *TransGlobal { m := TransGlobal{} dtmimp.MustUnmarshal(b, &m) logger.Debugf("creating trans in prepare") - // Payloads will be store in BinPayloads, Payloads is only used to Unmarshal - for _, p := range m.Payloads { - m.BinPayloads = append(m.BinPayloads, []byte(p)) - } - for _, d := range m.Steps { - if d["data"] != "" { - m.BinPayloads = append(m.BinPayloads, []byte(d["data"])) - } - } - m.Protocol = "http" - + m.setupPayloads() m.Ext.Headers = map[string]string{} if len(m.PassthroughHeaders) > 0 { for _, h := range m.PassthroughHeaders { @@ -103,6 +109,7 @@ func TransFromDtmRequest(ctx context.Context, c *dtmgpb.DtmRequest) *TransGlobal RetryInterval: o.RetryInterval, PassthroughHeaders: o.PassthroughHeaders, BranchHeaders: o.BranchHeaders, + RequestTimeout: o.RequestTimeout, }, }} if c.Steps != "" { diff --git a/dtmsvr/trans_status.go b/dtmsvr/trans_status.go index c45fc56..00c0c58 100644 --- a/dtmsvr/trans_status.go +++ b/dtmsvr/trans_status.go @@ -119,11 +119,13 @@ func (t *TransGlobal) getURLResult(url string, branchID, op string, branchPayloa if err != nil { return err } + conn := dtmgimp.MustGetGrpcConn(server, true) ctx := dtmgimp.TransInfo2Ctx(t.Gid, t.TransType, branchID, op, "") kvs := dtmgimp.Map2Kvs(t.Ext.Headers) kvs = append(kvs, dtmgimp.Map2Kvs(t.BranchHeaders)...) ctx = metadata.AppendToOutgoingContext(ctx, kvs...) + ctx = dtmgimp.RequestTimeoutNewContext(ctx, t.RequestTimeout) err = conn.Invoke(ctx, method, branchPayload, &[]byte{}) if err == nil { return nil diff --git a/dtmutil/consts.go b/dtmutil/consts.go index 9f1afe4..d8c4345 100644 --- a/dtmutil/consts.go +++ b/dtmutil/consts.go @@ -9,6 +9,8 @@ package dtmutil const ( // DefaultHTTPServer default url for http server. used by test and examples DefaultHTTPServer = "http://localhost:36789/api/dtmsvr" + // DefaultJrpcServer default url for http json-rpc server. used by test and examples + DefaultJrpcServer = "http://localhost:36789/api/json-rpc" // DefaultGrpcServer default url for grpc server. used by test and examples DefaultGrpcServer = "localhost:36790" ) diff --git a/test/busi/busi.pb.go b/test/busi/busi.pb.go index 09b8d85..ee64b5a 100644 --- a/test/busi/busi.pb.go +++ b/test/busi/busi.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.27.1 -// protoc v3.17.3 +// protoc v3.19.4 // source: test/busi/busi.proto package busi diff --git a/test/busi/busi_grpc.pb.go b/test/busi/busi_grpc.pb.go index 9235453..21828f1 100644 --- a/test/busi/busi_grpc.pb.go +++ b/test/busi/busi_grpc.pb.go @@ -1,4 +1,8 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.2.0 +// - protoc v3.19.4 +// source: test/busi/busi.proto package busi diff --git a/test/msg_jrpc_test.go b/test/msg_jrpc_test.go new file mode 100644 index 0000000..f94c2c8 --- /dev/null +++ b/test/msg_jrpc_test.go @@ -0,0 +1,71 @@ +/* + * Copyright (c) 2021 yedf. All rights reserved. + * Use of this source code is governed by a BSD-style + * license that can be found in the LICENSE file. + */ + +package test + +import ( + "testing" + + "github.com/dtm-labs/dtm/dtmcli" + "github.com/dtm-labs/dtm/dtmcli/dtmimp" + "github.com/dtm-labs/dtm/dtmutil" + "github.com/dtm-labs/dtm/test/busi" + "github.com/stretchr/testify/assert" +) + +func TestMsgJrpcNormal(t *testing.T) { + msg := genJrpcMsg(dtmimp.GetFuncName()) + msg.Submit() + assert.Equal(t, StatusSubmitted, getTransStatus(msg.Gid)) + waitTransProcessed(msg.Gid) + assert.Equal(t, []string{StatusSucceed, StatusSucceed}, getBranchesStatus(msg.Gid)) + assert.Equal(t, StatusSucceed, getTransStatus(msg.Gid)) +} + +func TestMsgJrpcRepeated(t *testing.T) { + msg := genJrpcMsg(dtmimp.GetFuncName()) + msg.Submit() + assert.Equal(t, StatusSubmitted, getTransStatus(msg.Gid)) + waitTransProcessed(msg.Gid) + assert.Equal(t, []string{StatusSucceed, StatusSucceed}, getBranchesStatus(msg.Gid)) + assert.Equal(t, StatusSucceed, getTransStatus(msg.Gid)) + err := msg.Submit() + assert.Error(t, err) +} +func TestMsgJprcAbnormal(t *testing.T) { + id := "no-use" + resp, err := dtmcli.GetRestyClient().R().SetBody("hello").Post(dtmutil.DefaultJrpcServer) + assert.Nil(t, err) + assert.Contains(t, resp.String(), "-32700") + + resp, err = dtmcli.GetRestyClient().R().SetBody(map[string]string{ + "jsonrpc": "1.0", + "method": "newGid", + "params": "", + "id": id, + }).Post(dtmutil.DefaultJrpcServer) + assert.Nil(t, err) + assert.Contains(t, resp.String(), "-32600") + + resp, err = dtmcli.GetRestyClient().R().SetBody(map[string]string{ + "jsonrpc": "2.0", + "method": "not-exists", + "params": "", + "id": id, + }).Post(dtmutil.DefaultJrpcServer) + assert.Nil(t, err) + assert.Contains(t, resp.String(), "-32601") +} + +func genJrpcMsg(gid string) *dtmcli.Msg { + req := busi.GenTransReq(30, false, false) + msg := dtmcli.NewMsg(dtmutil.DefaultJrpcServer, gid). + Add(busi.Busi+"/TransOut", &req). + Add(busi.Busi+"/TransIn", &req) + msg.QueryPrepared = busi.Busi + "/QueryPrepared" + msg.Protocol = "json-rpc" + return msg +} diff --git a/test/saga_grpc_test.go b/test/saga_grpc_test.go index ee181ce..94f1732 100644 --- a/test/saga_grpc_test.go +++ b/test/saga_grpc_test.go @@ -110,6 +110,17 @@ func TestSagaGrpcPassthroughHeadersYes(t *testing.T) { waitTransProcessed(gidYes) } +func TestSagaGrpcWithGlobalTransRequestTimeout(t *testing.T) { + gid := dtmimp.GetFuncName() + saga := dtmgrpc.NewSagaGrpc(dtmutil.DefaultGrpcServer, gid) + saga.WaitResult = true + saga.Add(busi.BusiGrpc+"/busi.Busi/TransOutHeaderNo", "", nil) + saga.WithGlobalTransRequestTimeout(6) + err := saga.Submit() + assert.Nil(t, err) + waitTransProcessed(gid) +} + func TestSagaGrpcCronPassthroughHeadersYes(t *testing.T) { gidYes := dtmimp.GetFuncName() sagaYes := dtmgrpc.NewSagaGrpc(dtmutil.DefaultGrpcServer, gidYes)