dtmgrpc

package module
v1.2.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 15, 2021 License: MIT Imports: 13 Imported by: 0

README

dtmgrpc

grpc client for dtm. code is synced from github.com/yedf/dtm/dtmgrpc

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var Dtm_ServiceDesc = grpc.ServiceDesc{
	ServiceName: "dtmgrpc.Dtm",
	HandlerType: (*DtmServer)(nil),
	Methods: []grpc.MethodDesc{
		{
			MethodName: "NewGid",
			Handler:    _Dtm_NewGid_Handler,
		},
		{
			MethodName: "Submit",
			Handler:    _Dtm_Submit_Handler,
		},
		{
			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",
}

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)

View Source 
var File_dtmgrpc_dtmgrpc_proto protoreflect.FileDescriptor

Functions

func GetGrpcConn 

func GetGrpcConn(grpcServer string) (conn *grpc.ClientConn, rerr error)

GetGrpcConn 1

func GetServerAndMethod 

func GetServerAndMethod(grpcURL string) (string, string)

GetServerAndMethod 将grpc的url分解为server和method

func GrpcClientLog 

func GrpcClientLog(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error

GrpcClientLog 打印grpc服务端的日志

func GrpcServerLog 

func GrpcServerLog(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error)

GrpcServerLog 打印grpc服务端的日志

func MustGenGid 

func MustGenGid(grpcServer string) string

MustGenGid 1

func MustGetGrpcConn 

func MustGetGrpcConn(grpcServer string) *grpc.ClientConn

MustGetGrpcConn 1

func RegisterDtmServer 

func RegisterDtmServer(s grpc.ServiceRegistrar, srv DtmServer)

func Result2Error 

func Result2Error(res interface{}, err error) error

Result2Error 将通用的result转成grpc的error

func SetCurrentDBType 

func SetCurrentDBType(dbType string)

SetCurrentDBType set the current db type

func TccGlobalTransaction 

func TccGlobalTransaction(dtm string, gid string, tccFunc TccGlobalFunc) (rerr error)

TccGlobalTransaction begin a tcc global transaction dtm dtm服务器地址 gid 全局事务id tccFunc tcc事务函数，里面会定义全局事务的分支

Types

type BranchBarrier 

type BranchBarrier struct {
	*dtmcli.BranchBarrier
}

BranchBarrier 子事务屏障

func BarrierFromGrpc 

func BarrierFromGrpc(in *BusiRequest) (*BranchBarrier, error)

BarrierFromGrpc 从BusiRequest生成一个Barrier

func (*BranchBarrier) Call 

func (bb *BranchBarrier) Call(tx dtmcli.Tx, busiCall dtmcli.BusiFunc) (rerr error)

Call 子事务屏障，详细介绍见 https://zhuanlan.zhihu.com/p/388444465 db: 本地数据库 transInfo: 事务信息 bisiCall: 业务函数，仅在必要时被调用 返回值: 如果发生悬挂，则busiCall不会被调用，直接返回错误 ErrFailure，全局事务尽早进行回滚 如果正常调用，重复调用，空补偿，返回的错误值为nil，正常往下进行

type BranchInfo 

type BranchInfo 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"`
	BranchID   string `protobuf:"bytes,3,opt,name=BranchID,proto3" json:"BranchID,omitempty"`
	BranchType string `protobuf:"bytes,4,opt,name=BranchType,proto3" json:"BranchType,omitempty"`
	// contains filtered or unexported fields
}

BranchInfo 事务分支信息

func (*BranchInfo) Descriptor deprecated 

func (*BranchInfo) Descriptor() ([]byte, []int)

Deprecated: Use BranchInfo.ProtoReflect.Descriptor instead.

func (*BranchInfo) GetBranchID 

func (x *BranchInfo) GetBranchID() string

func (*BranchInfo) GetBranchType 

func (x *BranchInfo) GetBranchType() string

func (*BranchInfo) GetGid 

func (x *BranchInfo) GetGid() string

func (*BranchInfo) GetTransType 

func (x *BranchInfo) GetTransType() string

func (*BranchInfo) ProtoMessage 

func (*BranchInfo) ProtoMessage()

func (*BranchInfo) ProtoReflect 

func (x *BranchInfo) ProtoReflect() protoreflect.Message

func (*BranchInfo) Reset 

func (x *BranchInfo) Reset()

func (*BranchInfo) String 

func (x *BranchInfo) String() string

type BusiReply 

type BusiReply struct {
	BusiData []byte `protobuf:"bytes,1,opt,name=BusiData,proto3" json:"BusiData,omitempty"`
	// contains filtered or unexported fields
}

BusiReply 业务响应数据

func (*BusiReply) Descriptor deprecated 

func (*BusiReply) Descriptor() ([]byte, []int)

Deprecated: Use BusiReply.ProtoReflect.Descriptor instead.

func (*BusiReply) GetBusiData 

func (x *BusiReply) GetBusiData() []byte

func (*BusiReply) ProtoMessage 

func (*BusiReply) ProtoMessage()

func (*BusiReply) ProtoReflect 

func (x *BusiReply) ProtoReflect() protoreflect.Message

func (*BusiReply) Reset 

func (x *BusiReply) Reset()

func (*BusiReply) String 

func (x *BusiReply) String() string

type BusiRequest 

type BusiRequest struct {
	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"`
	// contains filtered or unexported fields
}

BusiRequest 请求业务的数据，需要携带事务信息，便于业务进行幂等处理

func (*BusiRequest) Descriptor deprecated 

func (*BusiRequest) Descriptor() ([]byte, []int)

Deprecated: Use BusiRequest.ProtoReflect.Descriptor instead.

func (*BusiRequest) GetBusiData 

func (x *BusiRequest) GetBusiData() []byte

func (*BusiRequest) GetDtm 

func (x *BusiRequest) GetDtm() string

func (*BusiRequest) GetInfo 

func (x *BusiRequest) GetInfo() *BranchInfo

func (*BusiRequest) ProtoMessage 

func (*BusiRequest) ProtoMessage()

func (*BusiRequest) ProtoReflect 

func (x *BusiRequest) ProtoReflect() protoreflect.Message

func (*BusiRequest) Reset 

func (x *BusiRequest) Reset()

func (*BusiRequest) String 

func (x *BusiRequest) String() string

type DtmClient 

type DtmClient interface {
	NewGid(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*DtmGidReply, error)
	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)
}

DtmClient is the client API for Dtm service.

For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.

func MustGetDtmClient 

func MustGetDtmClient(grpcServer string) DtmClient

MustGetDtmClient 1

func NewDtmClient 

func NewDtmClient(cc grpc.ClientConnInterface) DtmClient

type DtmGidReply 

type DtmGidReply struct {
	Gid string `protobuf:"bytes,1,opt,name=Gid,proto3" json:"Gid,omitempty"`
	// contains filtered or unexported fields
}

func (*DtmGidReply) Descriptor deprecated 

func (*DtmGidReply) Descriptor() ([]byte, []int)

Deprecated: Use DtmGidReply.ProtoReflect.Descriptor instead.

func (*DtmGidReply) GetGid 

func (x *DtmGidReply) GetGid() string

func (*DtmGidReply) ProtoMessage 

func (*DtmGidReply) ProtoMessage()

func (*DtmGidReply) ProtoReflect 

func (x *DtmGidReply) ProtoReflect() protoreflect.Message

func (*DtmGidReply) Reset 

func (x *DtmGidReply) Reset()

func (*DtmGidReply) String 

func (x *DtmGidReply) String() string

type DtmRequest 

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 对于事务消息处于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"`
	// contains filtered or unexported fields
}

DtmRequest 发给dtm服务器的消息，响应为Emtpy，error == nil为成功，== Aborted 为失败 == 其他 可以重试

func (*DtmRequest) Descriptor deprecated 

func (*DtmRequest) Descriptor() ([]byte, []int)

Deprecated: Use DtmRequest.ProtoReflect.Descriptor instead.

func (*DtmRequest) GetData 

func (x *DtmRequest) GetData() string

func (*DtmRequest) GetGid 

func (x *DtmRequest) GetGid() string

func (*DtmRequest) GetQueryPrepared 

func (x *DtmRequest) GetQueryPrepared() string

func (*DtmRequest) GetTransType 

func (x *DtmRequest) GetTransType() string

func (*DtmRequest) GetWaitResult 

func (x *DtmRequest) GetWaitResult() bool

func (*DtmRequest) ProtoMessage 

func (*DtmRequest) ProtoMessage()

func (*DtmRequest) ProtoReflect 

func (x *DtmRequest) ProtoReflect() protoreflect.Message

func (*DtmRequest) Reset 

func (x *DtmRequest) Reset()

func (*DtmRequest) String 

func (x *DtmRequest) String() string

type DtmServer 

type DtmServer interface {
	NewGid(context.Context, *emptypb.Empty) (*DtmGidReply, error)
	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)
	// contains filtered or unexported methods
}

DtmServer is the server API for Dtm service. All implementations must embed UnimplementedDtmServer for forward compatibility

type DtmTccBranchRequest 

type DtmTccBranchRequest struct {
	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"`
	// contains filtered or unexported fields
}

func (*DtmTccBranchRequest) Descriptor deprecated 

func (*DtmTccBranchRequest) Descriptor() ([]byte, []int)

Deprecated: Use DtmTccBranchRequest.ProtoReflect.Descriptor instead.

func (*DtmTccBranchRequest) GetBusiData 

func (x *DtmTccBranchRequest) GetBusiData() string

func (*DtmTccBranchRequest) GetCancel 

func (x *DtmTccBranchRequest) GetCancel() string

func (*DtmTccBranchRequest) GetConfirm 

func (x *DtmTccBranchRequest) GetConfirm() string

func (*DtmTccBranchRequest) GetInfo 

func (x *DtmTccBranchRequest) GetInfo() *BranchInfo

func (*DtmTccBranchRequest) GetTry 

func (x *DtmTccBranchRequest) GetTry() string

func (*DtmTccBranchRequest) ProtoMessage 

func (*DtmTccBranchRequest) ProtoMessage()

func (*DtmTccBranchRequest) ProtoReflect 

func (x *DtmTccBranchRequest) ProtoReflect() protoreflect.Message

func (*DtmTccBranchRequest) Reset 

func (x *DtmTccBranchRequest) Reset()

func (*DtmTccBranchRequest) String 

func (x *DtmTccBranchRequest) String() string

type DtmXaBranchRequest 

type DtmXaBranchRequest struct {
	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"`
	// contains filtered or unexported fields
}

func (*DtmXaBranchRequest) Descriptor deprecated 

func (*DtmXaBranchRequest) Descriptor() ([]byte, []int)

Deprecated: Use DtmXaBranchRequest.ProtoReflect.Descriptor instead.

func (*DtmXaBranchRequest) GetBusiData 

func (x *DtmXaBranchRequest) GetBusiData() string

func (*DtmXaBranchRequest) GetInfo 

func (x *DtmXaBranchRequest) GetInfo() *BranchInfo

func (*DtmXaBranchRequest) GetNotify 

func (x *DtmXaBranchRequest) GetNotify() string

func (*DtmXaBranchRequest) ProtoMessage 

func (*DtmXaBranchRequest) ProtoMessage()

func (*DtmXaBranchRequest) ProtoReflect 

func (x *DtmXaBranchRequest) ProtoReflect() protoreflect.Message

func (*DtmXaBranchRequest) Reset 

func (x *DtmXaBranchRequest) Reset()

func (*DtmXaBranchRequest) String 

func (x *DtmXaBranchRequest) String() string

type MsgGrpc 

type MsgGrpc struct {
	dtmcli.TransBase
	Steps         []dtmcli.MsgStep `json:"steps"`
	QueryPrepared string           `json:"query_prepared"`
}

MsgGrpc reliable msg type

func NewMsgGrpc 

func NewMsgGrpc(server string, gid string) *MsgGrpc

NewMsgGrpc create new msg

func (*MsgGrpc) Add 

func (s *MsgGrpc) Add(action string, data []byte) *MsgGrpc

Add add a new step

func (*MsgGrpc) Prepare 

func (s *MsgGrpc) Prepare(queryPrepared string) error

Prepare prepare the msg

func (*MsgGrpc) Submit 

func (s *MsgGrpc) Submit() error

Submit submit the msg

type SagaGrpc 

type SagaGrpc struct {
	dtmcli.TransBase
	Steps []dtmcli.SagaStep `json:"steps"`
}

SagaGrpc struct of saga

func NewSaga 

func NewSaga(server string, gid string) *SagaGrpc

NewSaga create a saga

func (*SagaGrpc) Add 

func (s *SagaGrpc) Add(action string, compensate string, busiData []byte) *SagaGrpc

Add add a saga step

func (*SagaGrpc) Submit 

func (s *SagaGrpc) Submit() error

Submit submit the saga trans

type TccGlobalFunc 

type TccGlobalFunc func(tcc *TccGrpc) error

TccGlobalFunc type of global tcc call

type TccGrpc 

type TccGrpc struct {
	dtmcli.TransBase
}

TccGrpc struct of tcc

func TccFromRequest 

func TccFromRequest(br *BusiRequest) (*TccGrpc, error)

TccFromRequest tcc from request info

func (*TccGrpc) CallBranch 

func (t *TccGrpc) CallBranch(busiData []byte, tryURL string, confirmURL string, cancelURL string) (*BusiReply, error)

CallBranch call a tcc branch 函数首先注册子事务的所有分支，成功后调用try分支，返回try分支的调用结果

type UnimplementedDtmServer 

type UnimplementedDtmServer struct {
}

UnimplementedDtmServer must be embedded to have forward compatible implementations.

func (UnimplementedDtmServer) Abort 

func (UnimplementedDtmServer) Abort(context.Context, *DtmRequest) (*emptypb.Empty, error)

func (UnimplementedDtmServer) NewGid 

func (UnimplementedDtmServer) NewGid(context.Context, *emptypb.Empty) (*DtmGidReply, error)

func (UnimplementedDtmServer) Prepare 

func (UnimplementedDtmServer) Prepare(context.Context, *DtmRequest) (*emptypb.Empty, error)

func (UnimplementedDtmServer) RegisterTccBranch 

func (UnimplementedDtmServer) RegisterTccBranch(context.Context, *DtmTccBranchRequest) (*emptypb.Empty, error)

func (UnimplementedDtmServer) RegisterXaBranch 

func (UnimplementedDtmServer) RegisterXaBranch(context.Context, *DtmXaBranchRequest) (*emptypb.Empty, error)

func (UnimplementedDtmServer) Submit 

func (UnimplementedDtmServer) Submit(context.Context, *DtmRequest) (*emptypb.Empty, error)

type UnsafeDtmServer 

type UnsafeDtmServer interface {
	// contains filtered or unexported methods
}

UnsafeDtmServer may be embedded to opt out of forward compatibility for this service. Use of this interface is not recommended, as added methods to DtmServer will result in compilation errors.

type XaGrpc 

type XaGrpc struct {
	dtmcli.TransBase
}

XaGrpc xa transaction

func XaGrpcFromRequest 

func XaGrpcFromRequest(br *BusiRequest) (*XaGrpc, error)

XaGrpcFromRequest construct xa info from request

func (*XaGrpc) CallBranch 

func (x *XaGrpc) CallBranch(busiData []byte, url string) (*BusiReply, error)

CallBranch call a xa branch

type XaGrpcClient 

type XaGrpcClient struct {
	dtmcli.XaClientBase
}

XaGrpcClient xa client

func NewXaGrpcClient 

func NewXaGrpcClient(server string, mysqlConf map[string]string, notifyURL string) *XaGrpcClient

NewXaGrpcClient construct a xa client

func (*XaGrpcClient) HandleCallback 

func (xc *XaGrpcClient) HandleCallback(gid string, branchID string, action string) error

HandleCallback 处理commit/rollback的回调

func (*XaGrpcClient) XaGlobalTransaction 

func (xc *XaGrpcClient) XaGlobalTransaction(gid string, xaFunc XaGrpcGlobalFunc) (rerr error)

XaGlobalTransaction start a xa global transaction

func (*XaGrpcClient) XaLocalTransaction 

func (xc *XaGrpcClient) XaLocalTransaction(br *BusiRequest, xaFunc XaGrpcLocalFunc) (rerr error)

XaLocalTransaction start a xa local transaction

type XaGrpcGlobalFunc 

type XaGrpcGlobalFunc func(xa *XaGrpc) error

XaGrpcGlobalFunc type of xa global function

type XaGrpcLocalFunc 

type XaGrpcLocalFunc func(db *sql.DB, xa *XaGrpc) error

XaGrpcLocalFunc type of xa local function

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.