tsai
/
dtm
mirror of https://github.com/dtm-labs/dtm.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

grpc partial

pull/25/head
yedf2 5 years ago
parent
cab5f3e38c
commit
3136fc7945
9 changed files with 155 additions and 145 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 132
      dtmpb/dtmpb.pb.go
  2. 8
      dtmpb/dtmpb.proto
  3. 31
      dtmpb/type.go
  4. 3
      dtmsvr/api_grpc.go
  5. 58
      dtmsvr/trans.go
  6. 15
      dtmsvr/trans_msg.go
  7. 20
      dtmsvr/trans_saga.go
  8. 17
      dtmsvr/trans_tcc.go
  9. 16
      dtmsvr/trans_xa.go

132
dtmpb/dtmpb.pb.go
View File

@ -25,10 +25,11 @@ type DtmTransInfo struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Gid string `protobuf:"bytes,1,opt,name=Gid,proto3" json:"Gid,omitempty"`
TransType string `protobuf:"bytes,2,opt,name=TransType,proto3" json:"TransType,omitempty"`
BranchID string `protobuf:"bytes,3,opt,name=BranchID,proto3" json:"BranchID,omitempty"`
Dtm string `protobuf:"bytes,4,opt,name=Dtm,proto3" json:"Dtm,omitempty"`
Gid string `protobuf:"bytes,1,opt,name=Gid,proto3" json:"Gid,omitempty"`
TransType string `protobuf:"bytes,2,opt,name=TransType,proto3" json:"TransType,omitempty"`
BranchID string `protobuf:"bytes,3,opt,name=BranchID,proto3" json:"BranchID,omitempty"`
BranchType string `protobuf:"bytes,4,opt,name=BranchType,proto3" json:"BranchType,omitempty"`
Dtm string `protobuf:"bytes,5,opt,name=Dtm,proto3" json:"Dtm,omitempty"`
}
func (x *DtmTransInfo) Reset() {
@ -84,6 +85,13 @@ func (x *DtmTransInfo) GetBranchID() string {
return ""
}
func (x *DtmTransInfo) GetBranchType() string {
if x != nil {
return x.BranchType
}
return ""
}
func (x *DtmTransInfo) GetDtm() string {
if x != nil {
return x.Dtm
@ -100,8 +108,9 @@ type DtmRequest struct {
Gid string `protobuf:"bytes,1,opt,name=Gid,proto3" json:"Gid,omitempty"`
TransType string `protobuf:"bytes,2,opt,name=TransType,proto3" json:"TransType,omitempty"`
QueryPrepared string `protobuf:"bytes,3,opt,name=QueryPrepared,proto3" json:"QueryPrepared,omitempty"`
Extra map[string]string `protobuf:"bytes,4,rep,name=Extra,proto3" json:"Extra,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
Data string `protobuf:"bytes,5,opt,name=Data,proto3" json:"Data,omitempty"`
WaitResult bool `protobuf:"varint,4,opt,name=WaitResult,proto3" json:"WaitResult,omitempty"`
Extra map[string]string `protobuf:"bytes,5,rep,name=Extra,proto3" json:"Extra,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
Data string `protobuf:"bytes,6,opt,name=Data,proto3" json:"Data,omitempty"`
}
func (x *DtmRequest) Reset() {
@ -157,6 +166,13 @@ func (x *DtmRequest) GetQueryPrepared() string {
return ""
}
func (x *DtmRequest) GetWaitResult() bool {
if x != nil {
return x.WaitResult
}
return false
}
func (x *DtmRequest) GetExtra() map[string]string {
if x != nil {
return x.Extra
@ -351,59 +367,63 @@ var File_dtmpb_dtmpb_proto protoreflect.FileDescriptor
var file_dtmpb_dtmpb_proto_rawDesc = []byte{
0x0a, 0x11, 0x64, 0x74, 0x6d, 0x70, 0x62, 0x2f, 0x64, 0x74, 0x6d, 0x70, 0x62, 0x2e, 0x70, 0x72,
0x6f, 0x74, 0x6f, 0x12, 0x06, 0x64, 0x74, 0x6d, 0x63, 0x6c, 0x69, 0x22, 0x6c, 0x0a, 0x0c, 0x44,
0x74, 0x6d, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x49, 0x6e, 0x66, 0x6f, 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, 0x10, 0x0a, 0x03, 0x44, 0x74, 0x6d, 0x18, 0x04,
0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x44, 0x74, 0x6d, 0x22, 0xe5, 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, 0x24, 0x0a, 0x0d, 0x51, 0x75, 0x65, 0x72,
0x79, 0x50, 0x72, 0x65, 0x70, 0x61, 0x72, 0x65, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52,
0x0d, 0x51, 0x75, 0x65, 0x72, 0x79, 0x50, 0x72, 0x65, 0x70, 0x61, 0x72, 0x65, 0x64, 0x12, 0x33,
0x0a, 0x05, 0x45, 0x78, 0x74, 0x72, 0x61, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x1d, 0x2e,
0x64, 0x74, 0x6d, 0x63, 0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
0x74, 0x2e, 0x45, 0x78, 0x74, 0x72, 0x61, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x05, 0x45, 0x78,
0x74, 0x72, 0x61, 0x12, 0x12, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x18, 0x05, 0x20, 0x01, 0x28,
0x09, 0x52, 0x04, 0x44, 0x61, 0x74, 0x61, 0x1a, 0x38, 0x0a, 0x0a, 0x45, 0x78, 0x74, 0x72, 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, 0x22, 0x48, 0x0a, 0x08, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x1c, 0x0a,
0x6f, 0x74, 0x6f, 0x12, 0x06, 0x64, 0x74, 0x6d, 0x63, 0x6c, 0x69, 0x22, 0x8c, 0x01, 0x0a, 0x0c,
0x44, 0x74, 0x6d, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x49, 0x6e, 0x66, 0x6f, 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, 0x1e, 0x0a, 0x0a, 0x42, 0x72, 0x61, 0x6e,
0x63, 0x68, 0x54, 0x79, 0x70, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x42, 0x72,
0x61, 0x6e, 0x63, 0x68, 0x54, 0x79, 0x70, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x44, 0x74, 0x6d, 0x18,
0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x44, 0x74, 0x6d, 0x22, 0x85, 0x02, 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, 0x24, 0x0a, 0x0d, 0x51, 0x75, 0x65,
0x72, 0x79, 0x50, 0x72, 0x65, 0x70, 0x61, 0x72, 0x65, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09,
0x52, 0x0d, 0x51, 0x75, 0x65, 0x72, 0x79, 0x50, 0x72, 0x65, 0x70, 0x61, 0x72, 0x65, 0x64, 0x12,
0x1e, 0x0a, 0x0a, 0x57, 0x61, 0x69, 0x74, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x18, 0x04, 0x20,
0x01, 0x28, 0x08, 0x52, 0x0a, 0x57, 0x61, 0x69, 0x74, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12,
0x33, 0x0a, 0x05, 0x45, 0x78, 0x74, 0x72, 0x61, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x1d,
0x2e, 0x64, 0x74, 0x6d, 0x63, 0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65,
0x73, 0x74, 0x2e, 0x45, 0x78, 0x74, 0x72, 0x61, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x05, 0x45,
0x78, 0x74, 0x72, 0x61, 0x12, 0x12, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x18, 0x06, 0x20, 0x01,
0x28, 0x09, 0x52, 0x04, 0x44, 0x61, 0x74, 0x61, 0x1a, 0x38, 0x0a, 0x0a, 0x45, 0x78, 0x74, 0x72,
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, 0x22, 0x48, 0x0a, 0x08, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x1c,
0x0a, 0x09, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28,
0x09, 0x52, 0x09, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x1e, 0x0a, 0x0a,
0x44, 0x74, 0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
0x52, 0x0a, 0x44, 0x74, 0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0xc1, 0x01, 0x0a,
0x0b, 0x42, 0x75, 0x73, 0x69, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x28, 0x0a, 0x04,
0x69, 0x6e, 0x66, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x64, 0x74, 0x6d,
0x63, 0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x49, 0x6e, 0x66, 0x6f,
0x52, 0x04, 0x69, 0x6e, 0x66, 0x6f, 0x12, 0x34, 0x0a, 0x05, 0x45, 0x78, 0x74, 0x72, 0x61, 0x18,
0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x1e, 0x2e, 0x64, 0x74, 0x6d, 0x63, 0x6c, 0x69, 0x2e, 0x42,
0x75, 0x73, 0x69, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x2e, 0x45, 0x78, 0x74, 0x72, 0x61,
0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x05, 0x45, 0x78, 0x74, 0x72, 0x61, 0x12, 0x18, 0x0a, 0x07,
0x41, 0x70, 0x70, 0x44, 0x61, 0x74, 0x61, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x41,
0x70, 0x70, 0x44, 0x61, 0x74, 0x61, 0x1a, 0x38, 0x0a, 0x0a, 0x45, 0x78, 0x74, 0x72, 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,
0x22, 0x49, 0x0a, 0x09, 0x42, 0x75, 0x73, 0x69, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x1c, 0x0a,
0x09, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
0x52, 0x09, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x1e, 0x0a, 0x0a, 0x44,
0x74, 0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
0x0a, 0x44, 0x74, 0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0xc1, 0x01, 0x0a, 0x0b,
0x42, 0x75, 0x73, 0x69, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x28, 0x0a, 0x04, 0x69,
0x6e, 0x66, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x64, 0x74, 0x6d, 0x63,
0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x54, 0x72, 0x61, 0x6e, 0x73, 0x49, 0x6e, 0x66, 0x6f, 0x52,
0x04, 0x69, 0x6e, 0x66, 0x6f, 0x12, 0x34, 0x0a, 0x05, 0x45, 0x78, 0x74, 0x72, 0x61, 0x18, 0x02,
0x20, 0x03, 0x28, 0x0b, 0x32, 0x1e, 0x2e, 0x64, 0x74, 0x6d, 0x63, 0x6c, 0x69, 0x2e, 0x42, 0x75,
0x73, 0x69, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x2e, 0x45, 0x78, 0x74, 0x72, 0x61, 0x45,
0x6e, 0x74, 0x72, 0x79, 0x52, 0x05, 0x45, 0x78, 0x74, 0x72, 0x61, 0x12, 0x18, 0x0a, 0x07, 0x41,
0x70, 0x70, 0x44, 0x61, 0x74, 0x61, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x41, 0x70,
0x70, 0x44, 0x61, 0x74, 0x61, 0x1a, 0x38, 0x0a, 0x0a, 0x45, 0x78, 0x74, 0x72, 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, 0x22,
0x49, 0x0a, 0x09, 0x42, 0x75, 0x73, 0x69, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x1c, 0x0a, 0x09,
0x44, 0x74, 0x6d, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
0x09, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x1e, 0x0a, 0x0a, 0x44, 0x74,
0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a,
0x44, 0x74, 0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x32, 0x67, 0x0a, 0x03, 0x44, 0x74,
0x6d, 0x12, 0x2e, 0x0a, 0x04, 0x43, 0x61, 0x6c, 0x6c, 0x12, 0x12, 0x2e, 0x64, 0x74, 0x6d, 0x63,
0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x10, 0x2e,
0x64, 0x74, 0x6d, 0x63, 0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x22,
0x00, 0x12, 0x30, 0x0a, 0x06, 0x53, 0x75, 0x62, 0x6d, 0x69, 0x74, 0x12, 0x12, 0x2e, 0x64, 0x74,
0x6d, 0x63, 0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
0x10, 0x2e, 0x64, 0x74, 0x6d, 0x63, 0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x70, 0x6c,
0x79, 0x22, 0x00, 0x42, 0x1b, 0x5a, 0x19, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f,
0x6d, 0x2f, 0x79, 0x65, 0x64, 0x66, 0x2f, 0x64, 0x74, 0x6d, 0x2f, 0x64, 0x74, 0x6d, 0x70, 0x62,
0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
0x0a, 0x44, 0x74, 0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x32, 0x67, 0x0a, 0x03, 0x44,
0x74, 0x6d, 0x12, 0x2e, 0x0a, 0x04, 0x43, 0x61, 0x6c, 0x6c, 0x12, 0x12, 0x2e, 0x64, 0x74, 0x6d,
0x63, 0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x10,
0x2e, 0x64, 0x74, 0x6d, 0x63, 0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x70, 0x6c, 0x79,
0x22, 0x00, 0x12, 0x30, 0x0a, 0x06, 0x53, 0x75, 0x62, 0x6d, 0x69, 0x74, 0x12, 0x12, 0x2e, 0x64,
0x74, 0x6d, 0x63, 0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
0x1a, 0x10, 0x2e, 0x64, 0x74, 0x6d, 0x63, 0x6c, 0x69, 0x2e, 0x44, 0x74, 0x6d, 0x52, 0x65, 0x70,
0x6c, 0x79, 0x22, 0x00, 0x42, 0x1b, 0x5a, 0x19, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63,
0x6f, 0x6d, 0x2f, 0x79, 0x65, 0x64, 0x66, 0x2f, 0x64, 0x74, 0x6d, 0x2f, 0x64, 0x74, 0x6d, 0x70,
0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (

8
dtmpb/dtmpb.proto
View File

@ -15,7 +15,8 @@ message DtmTransInfo {
string Gid = 1;
string TransType = 2;
string BranchID = 3;
string Dtm = 4;
string BranchType = 4;
string Dtm = 5;
}
// The request message containing the user's name.
@ -23,8 +24,9 @@ message DtmRequest {
string Gid = 1;
string TransType = 2;
string QueryPrepared = 3;
map<string, string> Extra = 4;
string Data = 5;
bool WaitResult = 4;
map<string, string> Extra = 5;
string Data = 6;
}
// The response message containing the greetings

31
dtmpb/type.go
View File

@ -1,28 +1,43 @@
package dtmpb
import (
"strings"
"github.com/yedf/dtm/dtmcli"
grpc "google.golang.org/grpc"
)
var clients = map[string]DtmClient{}
var clients = map[string]*grpc.ClientConn{}
// GetDtmClient 1
func GetDtmClient(grpcServer string) (cli DtmClient, rerr error) {
// GetGrpcConn 1
func GetGrpcConn(grpcServer string) (conn *grpc.ClientConn, rerr error) {
if clients[grpcServer] == nil {
conn, err := grpc.Dial(grpcServer, grpc.WithInsecure(), grpc.WithBlock(), grpc.WithUnaryInterceptor(dtmcli.GrpcClientLog))
if err == nil {
clients[grpcServer] = NewDtmClient(conn)
clients[grpcServer] = conn
dtmcli.Logf("dtm client inited for %s", grpcServer)
}
}
cli = clients[grpcServer]
conn = clients[grpcServer]
return
}
// MustGetGrpcConn 1
func MustGetGrpcConn(grpcServer string) *grpc.ClientConn {
conn, err := GetGrpcConn(grpcServer)
dtmcli.E2P(err)
return conn
}
// MustGetDtmClient 1
func MustGetDtmClient(grpcServer string) DtmClient {
cli, err := GetDtmClient(grpcServer)
dtmcli.E2P(err)
return cli
return NewDtmClient(MustGetGrpcConn(grpcServer))
}
// GetServerAndMethod 将grpc的url分解为server和method
func GetServerAndMethod(grpcURL string) (string, string) {
fs := strings.Split(grpcURL, "/")
server := fs[0]
method := "/" + strings.Join(fs[1:], "/")
return server, method
}

3
dtmsvr/api_grpc.go
View File

@ -22,8 +22,7 @@ func (s *dtmServer) Call(ctx context.Context, in *pb.DtmRequest) (*pb.DtmReply,
}
func (s *dtmServer) Submit(ctx context.Context, in *pb.DtmRequest) (*pb.DtmReply, error) {
log.Printf("dtmServer Received: %v", in)
dynamicCallPb(ctx, in, in.Extra["BusiFunc"], []byte(in.Data))
svcSubmit(TransFromDtmRequest(in), in.WaitResult)
return &pb.DtmReply{DtmResult: "SUCCESS", DtmMessage: "ok"}, nil
}

58
dtmsvr/trans.go
View File

@ -1,7 +1,9 @@
package dtmsvr
import (
"context"
"fmt"
"strings"
"time"
"github.com/gin-gonic/gin"
@ -36,7 +38,6 @@ func (*TransGlobal) TableName() string {
type transProcessor interface {
GenBranches() []TransBranch
ProcessOnce(db *common.DB, branches []TransBranch)
ExecBranch(db *common.DB, branch *TransBranch)
}
func (t *TransGlobal) touch(db *common.DB, interval int64) *gorm.DB {
@ -147,15 +148,6 @@ func (t *TransGlobal) processInner(db *common.DB) (rerr error) {
return
}
func (t *TransGlobal) getBranchParams(branch *TransBranch) dtmcli.MS {
return dtmcli.MS{
"gid": t.Gid,
"trans_type": t.TransType,
"branch_id": branch.BranchID,
"branch_type": branch.BranchType,
}
}
func (t *TransGlobal) setNextCron(expireIn int64) []string {
t.NextCronInterval = expireIn
next := time.Now().Add(time.Duration(t.NextCronInterval) * time.Second)
@ -163,6 +155,48 @@ func (t *TransGlobal) setNextCron(expireIn int64) []string {
return []string{"next_cron_interval", "next_cron_time"}
}
func (t *TransGlobal) getBranchResult(branch *TransBranch) string {
if t.Protocol == "grpc" {
server, method := dtmpb.GetServerAndMethod(branch.URL)
conn := dtmpb.MustGetGrpcConn(server)
reply := dtmpb.BusiReply{}
err := conn.Invoke(context.Background(), method, &dtmpb.BusiRequest{
Info: &dtmpb.DtmTransInfo{
Gid: t.Gid,
TransType: t.TransType,
BranchID: branch.BranchID,
BranchType: branch.BranchType,
},
}, &reply)
e2p(err)
return reply.DtmResult
}
resp, err := dtmcli.RestyClient.R().SetBody(branch.Data).
SetQueryParams(dtmcli.MS{
"gid": t.Gid,
"trans_type": t.TransType,
"branch_id": branch.BranchID,
"branch_type": branch.BranchType,
}).
SetHeader("Content-type", "application/json").
Post(branch.URL)
e2p(err)
return resp.String()
}
func (t *TransGlobal) execBranch(db *common.DB, branch *TransBranch) {
body := t.getBranchResult(branch)
if strings.Contains(body, "SUCCESS") {
t.touch(db, config.TransCronInterval)
branch.changeStatus(db, "succeed")
} else if t.TransType == "saga" && branch.BranchType == "action" && strings.Contains(body, "FAILURE") {
t.touch(db, config.TransCronInterval)
branch.changeStatus(db, "failed")
} else {
panic(fmt.Errorf("unknown response: %s, will be retried", body))
}
}
func (t *TransGlobal) saveNew(db *common.DB) {
err := db.Transaction(func(db1 *gorm.DB) error {
db := &common.DB{DB: db1}
@ -205,15 +239,13 @@ func TransFromContext(c *gin.Context) *TransGlobal {
// TransFromDtmRequest TransFromContext
func TransFromDtmRequest(c *dtmpb.DtmRequest) *TransGlobal {
m := TransGlobal{
return &TransGlobal{
Gid: c.Gid,
TransType: c.TransType,
QueryPrepared: c.QueryPrepared,
Data: c.Data,
Protocol: "grpc",
}
m.Protocol = "http"
return &m
}
// TransFromDb construct trans from db

15
dtmsvr/trans_msg.go
View File

@ -1,7 +1,6 @@
package dtmsvr
import (
"fmt"
"strings"
"github.com/yedf/dtm/common"
@ -33,18 +32,6 @@ func (t *transMsgProcessor) GenBranches() []TransBranch {
return branches
}
func (t *transMsgProcessor) ExecBranch(db *common.DB, branch *TransBranch) {
resp, err := dtmcli.RestyClient.R().SetBody(branch.Data).SetQueryParams(t.getBranchParams(branch)).Post(branch.URL)
e2p(err)
body := resp.String()
if strings.Contains(body, "SUCCESS") {
branch.changeStatus(db, "succeed")
t.touch(db, config.TransCronInterval)
} else {
panic(fmt.Errorf("unknown response: %s, will be retried", body))
}
}
func (t *TransGlobal) mayQueryPrepared(db *common.DB) {
if t.Status != "prepared" {
return
@ -70,7 +57,7 @@ func (t *transMsgProcessor) ProcessOnce(db *common.DB, branches []TransBranch) {
if branch.BranchType != "action" || branch.Status != "prepared" {
continue
}
t.ExecBranch(db, branch)
t.execBranch(db, branch)
if branch.Status != "succeed" {
break
}

20
dtmsvr/trans_saga.go
View File

@ -2,7 +2,6 @@ package dtmsvr
import (
"fmt"
"strings"
"github.com/yedf/dtm/common"
"github.com/yedf/dtm/dtmcli"
@ -36,21 +35,6 @@ func (t *transSagaProcessor) GenBranches() []TransBranch {
return branches
}
func (t *transSagaProcessor) ExecBranch(db *common.DB, branch *TransBranch) {
resp, err := dtmcli.RestyClient.R().SetBody(branch.Data).SetQueryParams(t.getBranchParams(branch)).Post(branch.URL)
e2p(err)
body := resp.String()
if strings.Contains(body, "SUCCESS") {
t.touch(db, config.TransCronInterval)
branch.changeStatus(db, "succeed")
} else if branch.BranchType == "action" && strings.Contains(body, "FAILURE") {
t.touch(db, config.TransCronInterval)
branch.changeStatus(db, "failed")
} else {
panic(fmt.Errorf("unknown response: %s, will be retried", body))
}
}
func (t *transSagaProcessor) ProcessOnce(db *common.DB, branches []TransBranch) {
if t.Status == "failed" || t.Status == "succeed" {
return
@ -63,7 +47,7 @@ func (t *transSagaProcessor) ProcessOnce(db *common.DB, branches []TransBranch)
}
// 找到了一个非succeed的action
if branch.Status == "prepared" {
t.ExecBranch(db, branch)
t.execBranch(db, branch)
}
if branch.Status != "succeed" {
break
@ -81,7 +65,7 @@ func (t *transSagaProcessor) ProcessOnce(db *common.DB, branches []TransBranch)
if branch.BranchType != "compensate" || branch.Status != "prepared" {
continue
}
t.ExecBranch(db, branch)
t.execBranch(db, branch)
}
t.changeStatus(db, "failed")
}

17
dtmsvr/trans_tcc.go
View File

@ -1,9 +1,6 @@
package dtmsvr
import (
"fmt"
"strings"
"github.com/yedf/dtm/common"
"github.com/yedf/dtm/dtmcli"
)
@ -20,18 +17,6 @@ func (t *transTccProcessor) GenBranches() []TransBranch {
return []TransBranch{}
}
func (t *transTccProcessor) ExecBranch(db *common.DB, branch *TransBranch) {
resp, err := dtmcli.RestyClient.R().SetBody(branch.Data).SetHeader("Content-type", "application/json").SetQueryParams(t.getBranchParams(branch)).Post(branch.URL)
e2p(err)
body := resp.String()
if strings.Contains(body, "SUCCESS") {
t.touch(db, config.TransCronInterval)
branch.changeStatus(db, "succeed")
} else {
panic(fmt.Errorf("unknown response: %s, will be retried", body))
}
}
func (t *transTccProcessor) ProcessOnce(db *common.DB, branches []TransBranch) {
if t.Status == "succeed" || t.Status == "failed" {
return
@ -39,7 +24,7 @@ func (t *transTccProcessor) ProcessOnce(db *common.DB, branches []TransBranch) {
branchType := dtmcli.If(t.Status == "submitted", "confirm", "cancel").(string)
for current := len(branches) - 1; current >= 0; current-- {
if branches[current].BranchType == branchType && branches[current].Status == "prepared" {
t.ExecBranch(db, &branches[current])
t.execBranch(db, &branches[current])
}
}
t.changeStatus(db, dtmcli.If(t.Status == "submitted", "succeed", "failed").(string))

16
dtmsvr/trans_xa.go
View File

@ -1,9 +1,6 @@
package dtmsvr
import (
"fmt"
"strings"
"github.com/yedf/dtm/common"
"github.com/yedf/dtm/dtmcli"
)
@ -19,17 +16,6 @@ func init() {
func (t *transXaProcessor) GenBranches() []TransBranch {
return []TransBranch{}
}
func (t *transXaProcessor) ExecBranch(db *common.DB, branch *TransBranch) {
resp, err := dtmcli.RestyClient.R().SetQueryParams(t.getBranchParams(branch)).Post(branch.URL)
e2p(err)
body := resp.String()
if strings.Contains(body, "SUCCESS") {
t.touch(db, config.TransCronInterval)
branch.changeStatus(db, "succeed")
} else {
panic(fmt.Errorf("bad response: %s", body))
}
}
func (t *transXaProcessor) ProcessOnce(db *common.DB, branches []TransBranch) {
if t.Status == "succeed" {
@ -38,7 +24,7 @@ func (t *transXaProcessor) ProcessOnce(db *common.DB, branches []TransBranch) {
currentType := dtmcli.If(t.Status == "submitted", "commit", "rollback").(string)
for _, branch := range branches {
if branch.BranchType == currentType && branch.Status != "succeed" {
t.ExecBranch(db, &branch)
t.execBranch(db, &branch)
}
}
t.changeStatus(db, dtmcli.If(t.Status == "submitted", "succeed", "failed").(string))

Write Preview
Loading…
Cancel
Save