From e7518818986179fd3b1ce7bd47493ca3cd2c7b82 Mon Sep 17 00:00:00 2001 From: yedf2 <120050102@qq.com> Date: Tue, 10 Aug 2021 17:02:40 +0800 Subject: [PATCH] before fixing msg bug --- dtmcli/tcc.go | 1 - dtmcli/types.go | 13 +- dtmcli/xa.go | 18 +- dtmgrpc/dtmgrpc.pb.go | 565 +++++++++++++++++++++---------------- dtmgrpc/dtmgrpc.proto | 68 +++-- dtmgrpc/dtmgrpc_grpc.pb.go | 108 +++++++ dtmgrpc/saga.go | 6 +- dtmsvr/api_http.go | 3 +- dtmsvr/cron.go | 2 +- dtmsvr/trans.go | 4 +- examples/main_grpc.go | 12 +- 11 files changed, 515 insertions(+), 285 deletions(-) diff --git a/dtmcli/tcc.go b/dtmcli/tcc.go index 692c818..3d489a6 100644 --- a/dtmcli/tcc.go +++ b/dtmcli/tcc.go @@ -63,7 +63,6 @@ func (t *Tcc) CallBranch(body interface{}, tryURL string, confirmURL string, can "gid": t.Gid, "branch_id": branchID, "trans_type": "tcc", - "status": "prepared", "data": string(MustMarshal(body)), "try": tryURL, "confirm": confirmURL, diff --git a/dtmcli/types.go b/dtmcli/types.go index a5f03ad..64edcc9 100644 --- a/dtmcli/types.go +++ b/dtmcli/types.go @@ -60,12 +60,17 @@ type TransBase struct { WaitResult bool } +// NewTransBase 1 +func NewTransBase(dtm string, parentID string) *TransBase { + return &TransBase{ + IDGenerator: IDGenerator{parentID: parentID}, + Dtm: dtm, + } +} + // TransBaseFromQuery construct transaction info from request func TransBaseFromQuery(qs url.Values) *TransBase { - return &TransBase{ - IDGenerator: IDGenerator{parentID: qs.Get("branch_id")}, - Dtm: qs.Get("dtm"), - } + return NewTransBase(qs.Get("dtm"), qs.Get("branch_id")) } // CallDtm 调用dtm服务器，返回事务的状态 diff --git a/dtmcli/xa.go b/dtmcli/xa.go index 1d4ecfc..0967780 100644 --- a/dtmcli/xa.go +++ b/dtmcli/xa.go @@ -19,9 +19,9 @@ type XaRegisterCallback func(path string, xa *XaClient) // XaClient xa client type XaClient struct { - Server string - Conf map[string]string - CallbackURL string + Server string + Conf map[string]string + NotifyURL string } // Xa xa transaction @@ -40,13 +40,13 @@ func XaFromQuery(qs url.Values) (*Xa, error) { } // NewXaClient construct a xa client -func NewXaClient(server string, mysqlConf map[string]string, callbackURL string, register XaRegisterCallback) (*XaClient, error) { +func NewXaClient(server string, mysqlConf map[string]string, notifyURL string, register XaRegisterCallback) (*XaClient, error) { xa := &XaClient{ - Server: server, - Conf: mysqlConf, - CallbackURL: callbackURL, + Server: server, + Conf: mysqlConf, + NotifyURL: notifyURL, } - u, err := url.Parse(callbackURL) + u, err := url.Parse(notifyURL) if err != nil { return nil, err } @@ -103,7 +103,7 @@ func (xc *XaClient) XaLocalTransaction(qs url.Values, xaFunc XaLocalFunc) (ret i if rerr != nil { return } - rerr = xa.CallDtm(&M{"gid": xa.Gid, "branch_id": branchID, "trans_type": "xa", "status": "prepared", "url": xc.CallbackURL}, "registerXaBranch") + rerr = xa.CallDtm(&M{"gid": xa.Gid, "branch_id": branchID, "trans_type": "xa", "url": xc.NotifyURL}, "registerXaBranch") return } diff --git a/dtmgrpc/dtmgrpc.pb.go b/dtmgrpc/dtmgrpc.pb.go index 86df863..3df82ff 100644 --- a/dtmgrpc/dtmgrpc.pb.go +++ b/dtmgrpc/dtmgrpc.pb.go @@ -21,103 +21,26 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) -type DtmTransInfo struct { - state protoimpl.MessageState - 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"` - 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() { - *x = DtmTransInfo{} - if protoimpl.UnsafeEnabled { - mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[0] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) - } -} - -func (x *DtmTransInfo) String() string { - return protoimpl.X.MessageStringOf(x) -} - -func (*DtmTransInfo) ProtoMessage() {} - -func (x *DtmTransInfo) ProtoReflect() protoreflect.Message { - mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[0] - if protoimpl.UnsafeEnabled && x != nil { - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - if ms.LoadMessageInfo() == nil { - ms.StoreMessageInfo(mi) - } - return ms - } - return mi.MessageOf(x) -} - -// Deprecated: Use DtmTransInfo.ProtoReflect.Descriptor instead. -func (*DtmTransInfo) Descriptor() ([]byte, []int) { - return file_dtmgrpc_dtmgrpc_proto_rawDescGZIP(), []int{0} -} - -func (x *DtmTransInfo) GetGid() string { - if x != nil { - return x.Gid - } - return "" -} - -func (x *DtmTransInfo) GetTransType() string { - if x != nil { - return x.TransType - } - return "" -} - -func (x *DtmTransInfo) GetBranchID() string { - if x != nil { - return x.BranchID - } - return "" -} - -func (x *DtmTransInfo) GetBranchType() string { - if x != nil { - return x.BranchType - } - return "" -} - -func (x *DtmTransInfo) GetDtm() string { - if x != nil { - return x.Dtm - } - return "" -} - -// The request message containing the user's name. +// DtmRequest 发给dtm服务器的消息，响应为Emtpy，error == nil为成功，== Aborted 为失败 == 其他 可以重试 type DtmRequest struct { state protoimpl.MessageState 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"` - QueryPrepared string `protobuf:"bytes,3,opt,name=QueryPrepared,proto3" json:"QueryPrepared,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"` + 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 对于事务消息处于prepared状态过期，责护查询QueryPrepared + QueryPrepared string `protobuf:"bytes,3,opt,name=QueryPrepared,proto3" json:"QueryPrepared,omitempty"` + // WaitResult 设定这个值，Submit操作会等待dtm处理一次请求，可能在返回时，就可以把分布式事务完成 + WaitResult bool `protobuf:"varint,4,opt,name=WaitResult,proto3" json:"WaitResult,omitempty"` + // Data 包含saga、msg的子事务信息 + Data string `protobuf:"bytes,5,opt,name=Data,proto3" json:"Data,omitempty"` } func (x *DtmRequest) Reset() { *x = DtmRequest{} if protoimpl.UnsafeEnabled { - mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[1] + mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[0] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -130,7 +53,7 @@ func (x *DtmRequest) String() string { func (*DtmRequest) ProtoMessage() {} func (x *DtmRequest) ProtoReflect() protoreflect.Message { - mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[1] + mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[0] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -143,7 +66,7 @@ func (x *DtmRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use DtmRequest.ProtoReflect.Descriptor instead. func (*DtmRequest) Descriptor() ([]byte, []int) { - return file_dtmgrpc_dtmgrpc_proto_rawDescGZIP(), []int{1} + return file_dtmgrpc_dtmgrpc_proto_rawDescGZIP(), []int{0} } func (x *DtmRequest) GetGid() string { @@ -174,13 +97,6 @@ func (x *DtmRequest) GetWaitResult() bool { return false } -func (x *DtmRequest) GetExtra() map[string]string { - if x != nil { - return x.Extra - } - return nil -} - func (x *DtmRequest) GetData() string { if x != nil { return x.Data @@ -188,18 +104,92 @@ func (x *DtmRequest) GetData() string { return "" } -// The response message containing the greetings -type DtmReply struct { +// BranchInfo 事务分支信息 +type BranchInfo struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - DtmResult string `protobuf:"bytes,1,opt,name=DtmResult,proto3" json:"DtmResult,omitempty"` - DtmMessage string `protobuf:"bytes,2,opt,name=DtmMessage,proto3" json:"DtmMessage,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"` } -func (x *DtmReply) Reset() { - *x = DtmReply{} +func (x *BranchInfo) Reset() { + *x = BranchInfo{} + if protoimpl.UnsafeEnabled { + mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *BranchInfo) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*BranchInfo) ProtoMessage() {} + +func (x *BranchInfo) ProtoReflect() protoreflect.Message { + mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[1] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use BranchInfo.ProtoReflect.Descriptor instead. +func (*BranchInfo) Descriptor() ([]byte, []int) { + return file_dtmgrpc_dtmgrpc_proto_rawDescGZIP(), []int{1} +} + +func (x *BranchInfo) GetGid() string { + if x != nil { + return x.Gid + } + return "" +} + +func (x *BranchInfo) GetTransType() string { + if x != nil { + return x.TransType + } + return "" +} + +func (x *BranchInfo) GetBranchID() string { + if x != nil { + return x.BranchID + } + return "" +} + +func (x *BranchInfo) GetBranchType() string { + if x != nil { + return x.BranchType + } + return "" +} + +type DtmTccBranchRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Info *BranchInfo `protobuf:"bytes,1,opt,name=Info,proto3" json:"Info,omitempty"` + BusiData string `protobuf:"bytes,2,opt,name=BusiData,proto3" json:"BusiData,omitempty"` + Try string `protobuf:"bytes,3,opt,name=Try,proto3" json:"Try,omitempty"` + Confirm string `protobuf:"bytes,4,opt,name=Confirm,proto3" json:"Confirm,omitempty"` + Cancel string `protobuf:"bytes,5,opt,name=Cancel,proto3" json:"Cancel,omitempty"` +} + +func (x *DtmTccBranchRequest) Reset() { + *x = DtmTccBranchRequest{} if protoimpl.UnsafeEnabled { mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[2] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -207,13 +197,13 @@ func (x *DtmReply) Reset() { } } -func (x *DtmReply) String() string { +func (x *DtmTccBranchRequest) String() string { return protoimpl.X.MessageStringOf(x) } -func (*DtmReply) ProtoMessage() {} +func (*DtmTccBranchRequest) ProtoMessage() {} -func (x *DtmReply) ProtoReflect() protoreflect.Message { +func (x *DtmTccBranchRequest) ProtoReflect() protoreflect.Message { mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[2] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -225,40 +215,125 @@ func (x *DtmReply) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use DtmReply.ProtoReflect.Descriptor instead. -func (*DtmReply) Descriptor() ([]byte, []int) { +// Deprecated: Use DtmTccBranchRequest.ProtoReflect.Descriptor instead. +func (*DtmTccBranchRequest) Descriptor() ([]byte, []int) { return file_dtmgrpc_dtmgrpc_proto_rawDescGZIP(), []int{2} } -func (x *DtmReply) GetDtmResult() string { +func (x *DtmTccBranchRequest) GetInfo() *BranchInfo { if x != nil { - return x.DtmResult + return x.Info + } + return nil +} + +func (x *DtmTccBranchRequest) GetBusiData() string { + if x != nil { + return x.BusiData } return "" } -func (x *DtmReply) GetDtmMessage() string { +func (x *DtmTccBranchRequest) GetTry() string { if x != nil { - return x.DtmMessage + return x.Try } return "" } -// The request message containing the user's name. +func (x *DtmTccBranchRequest) GetConfirm() string { + if x != nil { + return x.Confirm + } + return "" +} + +func (x *DtmTccBranchRequest) GetCancel() string { + if x != nil { + return x.Cancel + } + return "" +} + +type DtmXaBranchRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Info *BranchInfo `protobuf:"bytes,1,opt,name=Info,proto3" json:"Info,omitempty"` + BusiData string `protobuf:"bytes,2,opt,name=BusiData,proto3" json:"BusiData,omitempty"` + // dtm通知业务提交和回滚的地址 + Notify string `protobuf:"bytes,3,opt,name=Notify,proto3" json:"Notify,omitempty"` +} + +func (x *DtmXaBranchRequest) Reset() { + *x = DtmXaBranchRequest{} + if protoimpl.UnsafeEnabled { + mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *DtmXaBranchRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*DtmXaBranchRequest) ProtoMessage() {} + +func (x *DtmXaBranchRequest) ProtoReflect() protoreflect.Message { + mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[3] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use DtmXaBranchRequest.ProtoReflect.Descriptor instead. +func (*DtmXaBranchRequest) Descriptor() ([]byte, []int) { + return file_dtmgrpc_dtmgrpc_proto_rawDescGZIP(), []int{3} +} + +func (x *DtmXaBranchRequest) GetInfo() *BranchInfo { + if x != nil { + return x.Info + } + return nil +} + +func (x *DtmXaBranchRequest) GetBusiData() string { + if x != nil { + return x.BusiData + } + return "" +} + +func (x *DtmXaBranchRequest) GetNotify() string { + if x != nil { + return x.Notify + } + return "" +} + +// BusiRequest 请求业务的数据，需要携带事务信息，便于业务进行幂等处理 type BusiRequest struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - Info *DtmTransInfo `protobuf:"bytes,1,opt,name=info,proto3" json:"info,omitempty"` - Extra map[string]string `protobuf:"bytes,2,rep,name=Extra,proto3" json:"Extra,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"` - AppData []byte `protobuf:"bytes,3,opt,name=AppData,proto3" json:"AppData,omitempty"` + Info *BranchInfo `protobuf:"bytes,1,opt,name=Info,proto3" json:"Info,omitempty"` + Dtm string `protobuf:"bytes,2,opt,name=Dtm,proto3" json:"Dtm,omitempty"` + BusiData []byte `protobuf:"bytes,3,opt,name=BusiData,proto3" json:"BusiData,omitempty"` } func (x *BusiRequest) Reset() { *x = BusiRequest{} if protoimpl.UnsafeEnabled { - mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[3] + mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -271,7 +346,7 @@ func (x *BusiRequest) String() string { func (*BusiRequest) ProtoMessage() {} func (x *BusiRequest) ProtoReflect() protoreflect.Message { - mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[3] + mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[4] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -284,44 +359,43 @@ func (x *BusiRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use BusiRequest.ProtoReflect.Descriptor instead. func (*BusiRequest) Descriptor() ([]byte, []int) { - return file_dtmgrpc_dtmgrpc_proto_rawDescGZIP(), []int{3} + return file_dtmgrpc_dtmgrpc_proto_rawDescGZIP(), []int{4} } -func (x *BusiRequest) GetInfo() *DtmTransInfo { +func (x *BusiRequest) GetInfo() *BranchInfo { if x != nil { return x.Info } return nil } -func (x *BusiRequest) GetExtra() map[string]string { +func (x *BusiRequest) GetDtm() string { if x != nil { - return x.Extra + return x.Dtm + } + return "" +} + +func (x *BusiRequest) GetBusiData() []byte { + if x != nil { + return x.BusiData } return nil } -func (x *BusiRequest) GetAppData() []byte { - if x != nil { - return x.AppData - } - return nil -} - -// The response message containing the greetings +// BusiReply 业务响应数据 type BusiReply struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - DtmResult string `protobuf:"bytes,1,opt,name=DtmResult,proto3" json:"DtmResult,omitempty"` - DtmMessage string `protobuf:"bytes,2,opt,name=DtmMessage,proto3" json:"DtmMessage,omitempty"` + BusiData []byte `protobuf:"bytes,1,opt,name=BusiData,proto3" json:"BusiData,omitempty"` } func (x *BusiReply) Reset() { *x = BusiReply{} if protoimpl.UnsafeEnabled { - mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[4] + mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -334,7 +408,7 @@ func (x *BusiReply) String() string { func (*BusiReply) ProtoMessage() {} func (x *BusiReply) ProtoReflect() protoreflect.Message { - mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[4] + mi := &file_dtmgrpc_dtmgrpc_proto_msgTypes[5] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -347,21 +421,14 @@ func (x *BusiReply) ProtoReflect() protoreflect.Message { // Deprecated: Use BusiReply.ProtoReflect.Descriptor instead. func (*BusiReply) Descriptor() ([]byte, []int) { - return file_dtmgrpc_dtmgrpc_proto_rawDescGZIP(), []int{4} + return file_dtmgrpc_dtmgrpc_proto_rawDescGZIP(), []int{5} } -func (x *BusiReply) GetDtmResult() string { +func (x *BusiReply) GetBusiData() []byte { if x != nil { - return x.DtmResult + return x.BusiData } - return "" -} - -func (x *BusiReply) GetDtmMessage() string { - if x != nil { - return x.DtmMessage - } - return "" + return nil } var File_dtmgrpc_dtmgrpc_proto protoreflect.FileDescriptor @@ -370,65 +437,74 @@ var file_dtmgrpc_dtmgrpc_proto_rawDesc = []byte{ 0x0a, 0x15, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x2f, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x07, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 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, 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, 0x86, 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, 0x34, 0x0a, 0x05, 0x45, 0x78, 0x74, 0x72, 0x61, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, - 0x32, 0x1e, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 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, - 0xc3, 0x01, 0x0a, 0x0b, 0x42, 0x75, 0x73, 0x69, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, - 0x29, 0x0a, 0x04, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, - 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x44, 0x74, 0x6d, 0x54, 0x72, 0x61, 0x6e, 0x73, - 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x04, 0x69, 0x6e, 0x66, 0x6f, 0x12, 0x35, 0x0a, 0x05, 0x45, 0x78, - 0x74, 0x72, 0x61, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x1f, 0x2e, 0x64, 0x74, 0x6d, 0x67, - 0x72, 0x70, 0x63, 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, 0x78, 0x0a, 0x03, 0x44, 0x74, 0x6d, 0x12, 0x37, 0x0a, 0x06, 0x53, 0x75, 0x62, 0x6d, 0x69, - 0x74, 0x12, 0x13, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 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, 0x72, 0x70, 0x63, 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, 0x42, 0x1d, 0x5a, 0x1b, 0x67, 0x69, - 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x79, 0x65, 0x64, 0x66, 0x2f, 0x64, 0x74, - 0x6d, 0x2f, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, - 0x33, + 0x66, 0x2f, 0x65, 0x6d, 0x70, 0x74, 0x79, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x96, 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, 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, 0x12, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, + 0x52, 0x04, 0x44, 0x61, 0x74, 0x61, 0x22, 0x78, 0x0a, 0x0a, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, + 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, + 0x22, 0x9e, 0x01, 0x0a, 0x13, 0x44, 0x74, 0x6d, 0x54, 0x63, 0x63, 0x42, 0x72, 0x61, 0x6e, 0x63, + 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x27, 0x0a, 0x04, 0x49, 0x6e, 0x66, 0x6f, + 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, + 0x2e, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x04, 0x49, 0x6e, 0x66, + 0x6f, 0x12, 0x1a, 0x0a, 0x08, 0x42, 0x75, 0x73, 0x69, 0x44, 0x61, 0x74, 0x61, 0x18, 0x02, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x08, 0x42, 0x75, 0x73, 0x69, 0x44, 0x61, 0x74, 0x61, 0x12, 0x10, 0x0a, + 0x03, 0x54, 0x72, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x54, 0x72, 0x79, 0x12, + 0x18, 0x0a, 0x07, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x72, 0x6d, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, + 0x52, 0x07, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x72, 0x6d, 0x12, 0x16, 0x0a, 0x06, 0x43, 0x61, 0x6e, + 0x63, 0x65, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x43, 0x61, 0x6e, 0x63, 0x65, + 0x6c, 0x22, 0x71, 0x0a, 0x12, 0x44, 0x74, 0x6d, 0x58, 0x61, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, + 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x27, 0x0a, 0x04, 0x49, 0x6e, 0x66, 0x6f, 0x18, + 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x2e, + 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x04, 0x49, 0x6e, 0x66, 0x6f, + 0x12, 0x1a, 0x0a, 0x08, 0x42, 0x75, 0x73, 0x69, 0x44, 0x61, 0x74, 0x61, 0x18, 0x02, 0x20, 0x01, + 0x28, 0x09, 0x52, 0x08, 0x42, 0x75, 0x73, 0x69, 0x44, 0x61, 0x74, 0x61, 0x12, 0x16, 0x0a, 0x06, + 0x4e, 0x6f, 0x74, 0x69, 0x66, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x4e, 0x6f, + 0x74, 0x69, 0x66, 0x79, 0x22, 0x64, 0x0a, 0x0b, 0x42, 0x75, 0x73, 0x69, 0x52, 0x65, 0x71, 0x75, + 0x65, 0x73, 0x74, 0x12, 0x27, 0x0a, 0x04, 0x49, 0x6e, 0x66, 0x6f, 0x18, 0x01, 0x20, 0x01, 0x28, + 0x0b, 0x32, 0x13, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x42, 0x72, 0x61, 0x6e, + 0x63, 0x68, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x04, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x10, 0x0a, 0x03, + 0x44, 0x74, 0x6d, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x44, 0x74, 0x6d, 0x12, 0x1a, + 0x0a, 0x08, 0x42, 0x75, 0x73, 0x69, 0x44, 0x61, 0x74, 0x61, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, + 0x52, 0x08, 0x42, 0x75, 0x73, 0x69, 0x44, 0x61, 0x74, 0x61, 0x22, 0x27, 0x0a, 0x09, 0x42, 0x75, + 0x73, 0x69, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x12, 0x1a, 0x0a, 0x08, 0x42, 0x75, 0x73, 0x69, 0x44, + 0x61, 0x74, 0x61, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x08, 0x42, 0x75, 0x73, 0x69, 0x44, + 0x61, 0x74, 0x61, 0x32, 0xc8, 0x02, 0x0a, 0x03, 0x44, 0x74, 0x6d, 0x12, 0x37, 0x0a, 0x06, 0x53, + 0x75, 0x62, 0x6d, 0x69, 0x74, 0x12, 0x13, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 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, 0x72, 0x70, 0x63, 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, 0x74, 0x6d, 0x67, 0x72, 0x70, + 0x63, 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, 0x4b, 0x0a, 0x11, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, + 0x65, 0x72, 0x54, 0x63, 0x63, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x12, 0x1c, 0x2e, 0x64, 0x74, + 0x6d, 0x67, 0x72, 0x70, 0x63, 0x2e, 0x44, 0x74, 0x6d, 0x54, 0x63, 0x63, 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, 0x12, 0x49, 0x0a, 0x10, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, 0x58, + 0x61, 0x42, 0x72, 0x61, 0x6e, 0x63, 0x68, 0x12, 0x1b, 0x2e, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, + 0x63, 0x2e, 0x44, 0x74, 0x6d, 0x58, 0x61, 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, 0x1d, + 0x5a, 0x1b, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x79, 0x65, 0x64, + 0x66, 0x2f, 0x64, 0x74, 0x6d, 0x2f, 0x64, 0x74, 0x6d, 0x67, 0x72, 0x70, 0x63, 0x62, 0x06, 0x70, + 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -443,27 +519,32 @@ func file_dtmgrpc_dtmgrpc_proto_rawDescGZIP() []byte { return file_dtmgrpc_dtmgrpc_proto_rawDescData } -var file_dtmgrpc_dtmgrpc_proto_msgTypes = make([]protoimpl.MessageInfo, 7) +var file_dtmgrpc_dtmgrpc_proto_msgTypes = make([]protoimpl.MessageInfo, 6) var file_dtmgrpc_dtmgrpc_proto_goTypes = []interface{}{ - (*DtmTransInfo)(nil), // 0: dtmgrpc.DtmTransInfo - (*DtmRequest)(nil), // 1: dtmgrpc.DtmRequest - (*DtmReply)(nil), // 2: dtmgrpc.DtmReply - (*BusiRequest)(nil), // 3: dtmgrpc.BusiRequest - (*BusiReply)(nil), // 4: dtmgrpc.BusiReply - nil, // 5: dtmgrpc.DtmRequest.ExtraEntry - nil, // 6: dtmgrpc.BusiRequest.ExtraEntry - (*emptypb.Empty)(nil), // 7: google.protobuf.Empty + (*DtmRequest)(nil), // 0: dtmgrpc.DtmRequest + (*BranchInfo)(nil), // 1: dtmgrpc.BranchInfo + (*DtmTccBranchRequest)(nil), // 2: dtmgrpc.DtmTccBranchRequest + (*DtmXaBranchRequest)(nil), // 3: dtmgrpc.DtmXaBranchRequest + (*BusiRequest)(nil), // 4: dtmgrpc.BusiRequest + (*BusiReply)(nil), // 5: dtmgrpc.BusiReply + (*emptypb.Empty)(nil), // 6: google.protobuf.Empty } var file_dtmgrpc_dtmgrpc_proto_depIdxs = []int32{ - 5, // 0: dtmgrpc.DtmRequest.Extra:type_name -> dtmgrpc.DtmRequest.ExtraEntry - 0, // 1: dtmgrpc.BusiRequest.info:type_name -> dtmgrpc.DtmTransInfo - 6, // 2: dtmgrpc.BusiRequest.Extra:type_name -> dtmgrpc.BusiRequest.ExtraEntry - 1, // 3: dtmgrpc.Dtm.Submit:input_type -> dtmgrpc.DtmRequest - 1, // 4: dtmgrpc.Dtm.Prepare:input_type -> dtmgrpc.DtmRequest - 7, // 5: dtmgrpc.Dtm.Submit:output_type -> google.protobuf.Empty - 7, // 6: dtmgrpc.Dtm.Prepare:output_type -> google.protobuf.Empty - 5, // [5:7] is the sub-list for method output_type - 3, // [3:5] is the sub-list for method input_type + 1, // 0: dtmgrpc.DtmTccBranchRequest.Info:type_name -> dtmgrpc.BranchInfo + 1, // 1: dtmgrpc.DtmXaBranchRequest.Info:type_name -> dtmgrpc.BranchInfo + 1, // 2: dtmgrpc.BusiRequest.Info:type_name -> dtmgrpc.BranchInfo + 0, // 3: dtmgrpc.Dtm.Submit:input_type -> dtmgrpc.DtmRequest + 0, // 4: dtmgrpc.Dtm.Prepare:input_type -> dtmgrpc.DtmRequest + 0, // 5: dtmgrpc.Dtm.Abort:input_type -> dtmgrpc.DtmRequest + 2, // 6: dtmgrpc.Dtm.RegisterTccBranch:input_type -> dtmgrpc.DtmTccBranchRequest + 3, // 7: dtmgrpc.Dtm.RegisterXaBranch:input_type -> dtmgrpc.DtmXaBranchRequest + 6, // 8: dtmgrpc.Dtm.Submit:output_type -> google.protobuf.Empty + 6, // 9: dtmgrpc.Dtm.Prepare:output_type -> google.protobuf.Empty + 6, // 10: dtmgrpc.Dtm.Abort:output_type -> google.protobuf.Empty + 6, // 11: dtmgrpc.Dtm.RegisterTccBranch:output_type -> google.protobuf.Empty + 6, // 12: dtmgrpc.Dtm.RegisterXaBranch:output_type -> google.protobuf.Empty + 8, // [8:13] is the sub-list for method output_type + 3, // [3:8] is the sub-list for method input_type 3, // [3:3] is the sub-list for extension type_name 3, // [3:3] is the sub-list for extension extendee 0, // [0:3] is the sub-list for field type_name @@ -476,18 +557,6 @@ func file_dtmgrpc_dtmgrpc_proto_init() { } if !protoimpl.UnsafeEnabled { file_dtmgrpc_dtmgrpc_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*DtmTransInfo); i { - case 0: - return &v.state - case 1: - return &v.sizeCache - case 2: - return &v.unknownFields - default: - return nil - } - } - file_dtmgrpc_dtmgrpc_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { switch v := v.(*DtmRequest); i { case 0: return &v.state @@ -499,8 +568,20 @@ func file_dtmgrpc_dtmgrpc_proto_init() { return nil } } + file_dtmgrpc_dtmgrpc_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*BranchInfo); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } file_dtmgrpc_dtmgrpc_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*DtmReply); i { + switch v := v.(*DtmTccBranchRequest); i { case 0: return &v.state case 1: @@ -512,7 +593,7 @@ func file_dtmgrpc_dtmgrpc_proto_init() { } } file_dtmgrpc_dtmgrpc_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*BusiRequest); i { + switch v := v.(*DtmXaBranchRequest); i { case 0: return &v.state case 1: @@ -524,6 +605,18 @@ func file_dtmgrpc_dtmgrpc_proto_init() { } } file_dtmgrpc_dtmgrpc_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*BusiRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_dtmgrpc_dtmgrpc_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} { switch v := v.(*BusiReply); i { case 0: return &v.state @@ -542,7 +635,7 @@ func file_dtmgrpc_dtmgrpc_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_dtmgrpc_dtmgrpc_proto_rawDesc, NumEnums: 0, - NumMessages: 7, + NumMessages: 6, NumExtensions: 0, NumServices: 1, }, diff --git a/dtmgrpc/dtmgrpc.proto b/dtmgrpc/dtmgrpc.proto index c2d9a8b..ab67223 100644 --- a/dtmgrpc/dtmgrpc.proto +++ b/dtmgrpc/dtmgrpc.proto @@ -9,42 +9,66 @@ package dtmgrpc; service Dtm { rpc Submit(DtmRequest) returns (google.protobuf.Empty) {} rpc Prepare(DtmRequest) returns (google.protobuf.Empty) {} - + rpc Abort(DtmRequest) returns (google.protobuf.Empty) {} + rpc RegisterTccBranch(DtmTccBranchRequest) returns (google.protobuf.Empty) {} + rpc RegisterXaBranch(DtmXaBranchRequest) returns (google.protobuf.Empty) {} } -message DtmTransInfo { +// DtmRequest 发给dtm服务器的消息，响应为Emtpy，error == nil为成功，== Aborted 为失败 == 其他 可以重试 +message DtmRequest { + string Gid = 1; + string TransType = 2; + // QueryPrepared 对于事务消息处于prepared状态过期，责护查询QueryPrepared + string QueryPrepared = 3; + // WaitResult 设定这个值，Submit操作会等待dtm处理一次请求，可能在返回时，就可以把分布式事务完成 + bool WaitResult = 4; + // Data 包含saga、msg的子事务信息 + string Data = 5; +} + +/* + "gid": t.Gid, + "branch_id": branchID, + "trans_type": "tcc", + "status": "prepared", + "data": string(MustMarshal(body)), + "try": tryURL, + "confirm": confirmURL, + "cancel": cancelURL, + +*/ + +// BranchInfo 事务分支信息 +message BranchInfo { string Gid = 1; string TransType = 2; string BranchID = 3; string BranchType = 4; - string Dtm = 5; } -// The request message containing the user's name. -message DtmRequest { - string Gid = 1; - string TransType = 2; - string QueryPrepared = 3; - bool WaitResult = 4; - map Extra = 5; - string Data = 6; +message DtmTccBranchRequest { + BranchInfo Info = 1; + string BusiData = 2; + string Try = 3; + string Confirm = 4; + string Cancel = 5; } -// The response message containing the greetings -message DtmReply { - string DtmResult = 1; - string DtmMessage = 2; +message DtmXaBranchRequest { + BranchInfo Info = 1; + string BusiData = 2; + // dtm通知业务提交和回滚的地址 + string Notify = 3; } -// The request message containing the user's name. +// BusiRequest 请求业务的数据，需要携带事务信息，便于业务进行幂等处理 message BusiRequest { - DtmTransInfo info = 1; - map Extra = 2; - bytes AppData = 3; + BranchInfo Info = 1; + string Dtm = 2; + bytes BusiData = 3; } -// The response message containing the greetings +// BusiReply 业务响应数据 message BusiReply { - string DtmResult = 1; - string DtmMessage = 2; + bytes BusiData = 1; } \ No newline at end of file diff --git a/dtmgrpc/dtmgrpc_grpc.pb.go b/dtmgrpc/dtmgrpc_grpc.pb.go index ce2efb5..d496f19 100644 --- a/dtmgrpc/dtmgrpc_grpc.pb.go +++ b/dtmgrpc/dtmgrpc_grpc.pb.go @@ -21,6 +21,9 @@ const _ = grpc.SupportPackageIsVersion7 type DtmClient interface { Submit(ctx context.Context, in *DtmRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) Prepare(ctx context.Context, in *DtmRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) + Abort(ctx context.Context, in *DtmRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) + RegisterTccBranch(ctx context.Context, in *DtmTccBranchRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) + RegisterXaBranch(ctx context.Context, in *DtmXaBranchRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) } type dtmClient struct { @@ -49,12 +52,42 @@ func (c *dtmClient) Prepare(ctx context.Context, in *DtmRequest, opts ...grpc.Ca return out, nil } +func (c *dtmClient) Abort(ctx context.Context, in *DtmRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) { + out := new(emptypb.Empty) + err := c.cc.Invoke(ctx, "/dtmgrpc.Dtm/Abort", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *dtmClient) RegisterTccBranch(ctx context.Context, in *DtmTccBranchRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) { + out := new(emptypb.Empty) + err := c.cc.Invoke(ctx, "/dtmgrpc.Dtm/RegisterTccBranch", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *dtmClient) RegisterXaBranch(ctx context.Context, in *DtmXaBranchRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) { + out := new(emptypb.Empty) + err := c.cc.Invoke(ctx, "/dtmgrpc.Dtm/RegisterXaBranch", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + // DtmServer is the server API for Dtm service. // All implementations must embed UnimplementedDtmServer // for forward compatibility type DtmServer interface { Submit(context.Context, *DtmRequest) (*emptypb.Empty, error) Prepare(context.Context, *DtmRequest) (*emptypb.Empty, error) + Abort(context.Context, *DtmRequest) (*emptypb.Empty, error) + RegisterTccBranch(context.Context, *DtmTccBranchRequest) (*emptypb.Empty, error) + RegisterXaBranch(context.Context, *DtmXaBranchRequest) (*emptypb.Empty, error) mustEmbedUnimplementedDtmServer() } @@ -68,6 +101,15 @@ func (UnimplementedDtmServer) Submit(context.Context, *DtmRequest) (*emptypb.Emp func (UnimplementedDtmServer) Prepare(context.Context, *DtmRequest) (*emptypb.Empty, error) { return nil, status.Errorf(codes.Unimplemented, "method Prepare not implemented") } +func (UnimplementedDtmServer) Abort(context.Context, *DtmRequest) (*emptypb.Empty, error) { + return nil, status.Errorf(codes.Unimplemented, "method Abort not implemented") +} +func (UnimplementedDtmServer) RegisterTccBranch(context.Context, *DtmTccBranchRequest) (*emptypb.Empty, error) { + return nil, status.Errorf(codes.Unimplemented, "method RegisterTccBranch not implemented") +} +func (UnimplementedDtmServer) RegisterXaBranch(context.Context, *DtmXaBranchRequest) (*emptypb.Empty, error) { + return nil, status.Errorf(codes.Unimplemented, "method RegisterXaBranch not implemented") +} func (UnimplementedDtmServer) mustEmbedUnimplementedDtmServer() {} // UnsafeDtmServer may be embedded to opt out of forward compatibility for this service. @@ -117,6 +159,60 @@ func _Dtm_Prepare_Handler(srv interface{}, ctx context.Context, dec func(interfa return interceptor(ctx, in, info, handler) } +func _Dtm_Abort_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(DtmRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(DtmServer).Abort(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/dtmgrpc.Dtm/Abort", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(DtmServer).Abort(ctx, req.(*DtmRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Dtm_RegisterTccBranch_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(DtmTccBranchRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(DtmServer).RegisterTccBranch(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/dtmgrpc.Dtm/RegisterTccBranch", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(DtmServer).RegisterTccBranch(ctx, req.(*DtmTccBranchRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Dtm_RegisterXaBranch_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(DtmXaBranchRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(DtmServer).RegisterXaBranch(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/dtmgrpc.Dtm/RegisterXaBranch", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(DtmServer).RegisterXaBranch(ctx, req.(*DtmXaBranchRequest)) + } + return interceptor(ctx, in, info, handler) +} + // Dtm_ServiceDesc is the grpc.ServiceDesc for Dtm service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -132,6 +228,18 @@ var Dtm_ServiceDesc = grpc.ServiceDesc{ MethodName: "Prepare", Handler: _Dtm_Prepare_Handler, }, + { + MethodName: "Abort", + Handler: _Dtm_Abort_Handler, + }, + { + MethodName: "RegisterTccBranch", + Handler: _Dtm_RegisterTccBranch_Handler, + }, + { + MethodName: "RegisterXaBranch", + Handler: _Dtm_RegisterXaBranch_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "dtmgrpc/dtmgrpc.proto", diff --git a/dtmgrpc/saga.go b/dtmgrpc/saga.go index 1326627..7f20a82 100644 --- a/dtmgrpc/saga.go +++ b/dtmgrpc/saga.go @@ -26,12 +26,12 @@ func NewSaga(server string, gid string) *SagaGrpc { } // Add add a saga step -func (s *SagaGrpc) Add(action string, compensate string, appData []byte) *SagaGrpc { - dtmcli.Logf("saga %s Add %s %s %v", s.SagaData.Gid, action, compensate, string(appData)) +func (s *SagaGrpc) Add(action string, compensate string, busiData []byte) *SagaGrpc { + dtmcli.Logf("saga %s Add %s %s %v", s.SagaData.Gid, action, compensate, string(busiData)) step := dtmcli.SagaStep{ Action: action, Compensate: compensate, - Data: string(appData), + Data: string(busiData), } s.Steps = append(s.Steps, step) return s diff --git a/dtmsvr/api_http.go b/dtmsvr/api_http.go index 29a9220..edbff98 100644 --- a/dtmsvr/api_http.go +++ b/dtmsvr/api_http.go @@ -50,6 +50,7 @@ func registerXaBranch(c *gin.Context) (interface{}, error) { branch := TransBranch{} err := c.BindJSON(&branch) e2p(err) + branch.Status = "prepared" db := dbGet() dbt := TransFromDb(db, branch.Gid) if dbt.Status != "prepared" { @@ -74,7 +75,7 @@ func registerTccBranch(c *gin.Context) (interface{}, error) { branch := TransBranch{ Gid: data["gid"], BranchID: data["branch_id"], - Status: data["status"], + Status: "prepared", Data: data["data"], } db := dbGet() diff --git a/dtmsvr/cron.go b/dtmsvr/cron.go index 2ff1f1d..852461f 100644 --- a/dtmsvr/cron.go +++ b/dtmsvr/cron.go @@ -54,7 +54,7 @@ func lockOneTrans(expireIn time.Duration) *TransGlobal { func handlePanic(perr *error) { if err := recover(); err != nil { - dtmcli.LogRedf("----panic %v handlered\n%s", err, string(debug.Stack())) + dtmcli.LogRedf("----recovered panic %v\n%s", err, string(debug.Stack())) if perr != nil { *perr = fmt.Errorf("dtm panic: %v", err) } diff --git a/dtmsvr/trans.go b/dtmsvr/trans.go index bf20030..7e7f9c4 100644 --- a/dtmsvr/trans.go +++ b/dtmsvr/trans.go @@ -164,13 +164,13 @@ func (t *TransGlobal) getURLResult(url string, branchID, branchType string, bran server, method := dtmgrpc.GetServerAndMethod(url) conn := dtmgrpc.MustGetGrpcConn(server) err := conn.Invoke(context.Background(), method, &dtmgrpc.BusiRequest{ - Info: &dtmgrpc.DtmTransInfo{ + Info: &dtmgrpc.BranchInfo{ Gid: t.Gid, TransType: t.TransType, BranchID: branchID, BranchType: branchType, }, - AppData: []byte(branchData), + BusiData: []byte(branchData), }, &emptypb.Empty{}) if err == nil { return "SUCCESS" diff --git a/examples/main_grpc.go b/examples/main_grpc.go index d92dd11..2e7a928 100644 --- a/examples/main_grpc.go +++ b/examples/main_grpc.go @@ -61,36 +61,36 @@ func (s *busiServer) CanSubmit(ctx context.Context, in *dtmgrpc.BusiRequest) (*e func (s *busiServer) TransIn(ctx context.Context, in *dtmgrpc.BusiRequest) (*emptypb.Empty, error) { req := TransReq{} - dtmcli.MustUnmarshal(in.AppData, &req) + dtmcli.MustUnmarshal(in.BusiData, &req) return &emptypb.Empty{}, handleGrpcBusiness(in, MainSwitch.TransInResult.Fetch(), req.TransInResult, dtmcli.GetFuncName()) } func (s *busiServer) TransOut(ctx context.Context, in *dtmgrpc.BusiRequest) (*emptypb.Empty, error) { req := TransReq{} - dtmcli.MustUnmarshal(in.AppData, &req) + dtmcli.MustUnmarshal(in.BusiData, &req) return &emptypb.Empty{}, handleGrpcBusiness(in, MainSwitch.TransOutResult.Fetch(), req.TransOutResult, dtmcli.GetFuncName()) } func (s *busiServer) TransInRevert(ctx context.Context, in *dtmgrpc.BusiRequest) (*emptypb.Empty, error) { req := TransReq{} - dtmcli.MustUnmarshal(in.AppData, &req) + dtmcli.MustUnmarshal(in.BusiData, &req) return &emptypb.Empty{}, handleGrpcBusiness(in, MainSwitch.TransInRevertResult.Fetch(), "", dtmcli.GetFuncName()) } func (s *busiServer) TransOutRevert(ctx context.Context, in *dtmgrpc.BusiRequest) (*emptypb.Empty, error) { req := TransReq{} - dtmcli.MustUnmarshal(in.AppData, &req) + dtmcli.MustUnmarshal(in.BusiData, &req) return &emptypb.Empty{}, handleGrpcBusiness(in, MainSwitch.TransOutRevertResult.Fetch(), "", dtmcli.GetFuncName()) } func (s *busiServer) TransInConfirm(ctx context.Context, in *dtmgrpc.BusiRequest) (*emptypb.Empty, error) { req := TransReq{} - dtmcli.MustUnmarshal(in.AppData, &req) + dtmcli.MustUnmarshal(in.BusiData, &req) return &emptypb.Empty{}, handleGrpcBusiness(in, MainSwitch.TransInConfirmResult.Fetch(), "", dtmcli.GetFuncName()) } func (s *busiServer) TransOutConfirm(ctx context.Context, in *dtmgrpc.BusiRequest) (*emptypb.Empty, error) { req := TransReq{} - dtmcli.MustUnmarshal(in.AppData, &req) + dtmcli.MustUnmarshal(in.BusiData, &req) return &emptypb.Empty{}, handleGrpcBusiness(in, MainSwitch.TransOutConfirmResult.Fetch(), "", dtmcli.GetFuncName()) }