dtmimp

package
v0.0.0-...-431ae09 Latest Latest
Warning

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

 Go to latest Published: Apr 11, 2023 License: BSD-3-Clause Imports: 16 Imported by: 0

README

Notice

Please donot use this package, and this package should only be used in dtm internally. The interfaces are not stable, and package name has postfix "imp"

Documentation

Index

Constants

View Source 
const (
	// ResultFailure for result of a trans/trans branch
	// Same as HTTP status 409 and GRPC code 10
	ResultFailure = "FAILURE"
	// ResultSuccess for result of a trans/trans branch
	// Same as HTTP status 200 and GRPC code 0
	ResultSuccess = "SUCCESS"
	// ResultOngoing for result of a trans/trans branch
	// Same as HTTP status 425 and GRPC code 9
	ResultOngoing = "ONGOING"

	// OpTry branch type for TCC
	OpTry = "try"
	// OpConfirm branch type for TCC
	OpConfirm = "confirm"
	// OpCancel branch type for TCC
	OpCancel = "cancel"
	// OpAction branch type for message, SAGA, XA
	OpAction = "action"
	// OpCompensate branch type for SAGA
	OpCompensate = "compensate"
	// OpCommit branch type for XA
	OpCommit = "commit"
	// OpRollback branch type for XA
	OpRollback = "rollback"

	// DBTypeMysql const for driver mysql
	DBTypeMysql = "mysql"
	// DBTypePostgres const for driver postgres
	DBTypePostgres = "postgres"
	// DBTypeRedis const for driver redis
	DBTypeRedis = "redis"
	// Jrpc const for json-rpc
	Jrpc = "json-rpc"
	// JrpcCodeFailure const for json-rpc failure
	JrpcCodeFailure = -32901

	// JrpcCodeOngoing const for json-rpc ongoing
	JrpcCodeOngoing = -32902

	// MsgDoBranch0 const for DoAndSubmit barrier branch
	MsgDoBranch0 = "00"
	// MsgDoBarrier1 const for DoAndSubmit barrier barrierID
	MsgDoBarrier1 = "01"
	// MsgDoOp const for DoAndSubmit barrier op
	MsgDoOp = "msg"
	//MsgTopicPrefix const for Add topic msg
	MsgTopicPrefix = "topic://"

	// XaBarrier1 const for xa barrier id
	XaBarrier1 = "01"

	// ProtocolGRPC const for protocol grpc
	ProtocolGRPC = "grpc"
	// ProtocolHTTP const for protocol http
	ProtocolHTTP = "http"
)

Variables

View Source 
var BarrierTableName = "dtm_barrier.barrier"

BarrierTableName the table name of barrier table

View Source 
var ErrDuplicated = errors.New("DUPLICATED")

ErrDuplicated error of DUPLICATED for only msg if QueryPrepared executed before call. then DoAndSubmit return this error

View Source 
var ErrFailure = errors.New("FAILURE")

ErrFailure error of FAILURE

View Source 
var ErrOngoing = errors.New("ONGOING")

ErrOngoing error of ONGOING

View Source 
var FatalIfError = logger.FatalIfError

FatalIfError fatal if error is not nil Deprecated: use logger.FatalIfError

View Source 
var LogIfFatalf = logger.FatalfIf

LogIfFatalf fatal if cond is true Deprecated: use logger.FatalfIf

View Source 
var LogRedf = logger.Errorf

LogRedf an alias of Errorf Deprecated: use logger.Errorf

View Source 
var Logf = logger.Infof

Logf an alias of Infof Deprecated: use logger.Errorf

View Source 
var MapFailure = map[string]interface{}{"dtm_result": ResultFailure}

MapFailure HTTP result of FAILURE

View Source 
var MapSuccess = map[string]interface{}{"dtm_result": ResultSuccess}

MapSuccess HTTP result of SUCCESS

Functions

func AddRestyMiddlewares 

func AddRestyMiddlewares(client *resty.Client)

AddRestyMiddlewares will add the middlewares used by dtm

func AsError 

func AsError(x interface{}) error

AsError wrap a panic value as an error

func CatchP 

func CatchP(f func()) (rerr error)

CatchP catch panic to error

func DBExec 

func DBExec(dbType string, db DB, sql string, values ...interface{}) (affected int64, rerr error)

DBExec use raw db to exec

func DeferDo 

func DeferDo(rerr *error, success func() error, fail func() error)

DeferDo a common defer do used in dtmcli/dtmgrpc

func E2P 

func E2P(err error)

E2P error to panic

func Escape 

func Escape(input string) string

Escape solve CodeQL reported problem

func EscapeGet 

func EscapeGet(qs url.Values, key string) string

EscapeGet escape get

func GetCurrentDBType 

func GetCurrentDBType() string

GetCurrentDBType get currentDBType

func GetDsn 

func GetDsn(conf DBConf) string

GetDsn get dsn from map config

func GetFuncName 

func GetFuncName() string

GetFuncName get current call func name

func GetRestyClient2 

func GetRestyClient2(timeout time.Duration) *resty.Client

GetRestyClient2 will return a resty client with timeout set

func If 

func If(condition bool, trueObj interface{}, falseObj interface{}) interface{}

If ternary operator

func InsertBarrier 

func InsertBarrier(tx DB, transType string, gid string, branchID string, op string, barrierID string, reason string, dbType string, barrierTableName string) (int64, error)

InsertBarrier insert a record to barrier

func MayReplaceLocalhost 

func MayReplaceLocalhost(host string) string

MayReplaceLocalhost when run in docker compose, change localhost to host.docker.internal for accessing host network

func MustAtoi 

func MustAtoi(s string) int

MustAtoi is string to int

func MustMarshal 

func MustMarshal(v interface{}) []byte

MustMarshal checked version for marshal

func MustMarshalString 

func MustMarshalString(v interface{}) string

MustMarshalString string version of MustMarshal

func MustRemarshal 

func MustRemarshal(from interface{}, to interface{})

MustRemarshal marshal and unmarshal, and check error

func MustUnmarshal 

func MustUnmarshal(b []byte, obj interface{})

MustUnmarshal checked version for unmarshal

func MustUnmarshalString 

func MustUnmarshalString(s string, obj interface{})

MustUnmarshalString string version of MustUnmarshal

func OrString 

func OrString(ss ...string) string

OrString return the first not empty string

func P2E 

func P2E(perr *error)

P2E panic to error

func PanicIf 

func PanicIf(cond bool, err error)

PanicIf name is clear

func PooledDB 

func PooledDB(conf DBConf) (*sql.DB, error)

PooledDB get pooled sql.DB

func RespAsErrorByJSONRPC 

func RespAsErrorByJSONRPC(resp *resty.Response) error

RespAsErrorByJSONRPC translate json rpc resty response to error

func SetCurrentDBType 

func SetCurrentDBType(dbType string)

SetCurrentDBType set currentDBType

func StandaloneDB 

func StandaloneDB(conf DBConf) (*sql.DB, error)

StandaloneDB get a standalone db instance

func TransCallDtm 

func TransCallDtm(tb *TransBase, operation string) error

TransCallDtm is the short call for TransCallDtmExt

func TransCallDtmExt 

func TransCallDtmExt(tb *TransBase, body interface{}, operation string) (*resty.Response, error)

TransCallDtmExt TransBase call dtm

func TransRegisterBranch 

func TransRegisterBranch(tb *TransBase, added map[string]string, operation string) error

TransRegisterBranch TransBase register a branch to dtm

func TransRequestBranch 

func TransRequestBranch(t *TransBase, method string, body interface{}, branchID string, op string, url string) (*resty.Response, error)

TransRequestBranch TransBase request branch result

func XaClose 

func XaClose(db *sql.DB)

XaClose will log and close the db

func XaDB 

func XaDB(conf DBConf) (*sql.DB, error)

XaDB return a standalone db instance for xa

func XaHandleGlobalTrans 

func XaHandleGlobalTrans(xa *TransBase, callDtm func(string) error, callBusi func() error) (rerr error)

XaHandleGlobalTrans http/grpc GlobalTransaction shared func

func XaHandleLocalTrans 

func XaHandleLocalTrans(xa *TransBase, dbConf DBConf, cb func(*sql.DB) error) (rerr error)

XaHandleLocalTrans public handler of LocalTransaction via http/grpc

func XaHandlePhase2 

func XaHandlePhase2(gid string, dbConf DBConf, branchID string, op string) error

XaHandlePhase2 Handle the callback of commit/rollback

Types

type BranchIDGen 

type BranchIDGen struct {
	BranchID string
	// contains filtered or unexported fields
}

BranchIDGen used to generate a sub branch id

func (*BranchIDGen) CurrentSubBranchID 

func (g *BranchIDGen) CurrentSubBranchID() string

CurrentSubBranchID return current branchID

func (*BranchIDGen) NewSubBranchID 

func (g *BranchIDGen) NewSubBranchID() string

NewSubBranchID generate a sub branch id

type DB 

type DB interface {
	Exec(query string, args ...interface{}) (sql.Result, error)
	QueryRow(query string, args ...interface{}) *sql.Row
}

DB inteface of dtmcli db

type DBConf 

type DBConf struct {
	Driver   string `yaml:"Driver"`
	Host     string `yaml:"Host"`
	Port     int64  `yaml:"Port"`
	User     string `yaml:"User"`
	Password string `yaml:"Password"`
	Db       string `yaml:"Db"`
	Schema   string `yaml:"Schema"`
}

DBConf defines db config

type DBSpecial 

type DBSpecial interface {
	GetPlaceHoldSQL(sql string) string
	GetInsertIgnoreTemplate(tableAndValues string, pgConstraint string) string
	GetXaSQL(command string, xid string) string
}

DBSpecial db specific operations

func GetDBSpecial 

func GetDBSpecial(dbType string) DBSpecial

GetDBSpecial get DBSpecial for currentDBType

type TransBase 

type TransBase struct {
	Gid        string `json:"gid"` //  NOTE: unique in storage, can customize the generation rules instead of using server-side generation, it will help with the tracking
	TransType  string `json:"trans_type"`
	Dtm        string `json:"-"`
	CustomData string `json:"custom_data,omitempty"` // nosql data persistence
	TransOptions
	Context context.Context `json:"-" gorm:"-"`

	Steps       []map[string]string `json:"steps,omitempty"`    // use in MSG/SAGA
	Payloads    []string            `json:"payloads,omitempty"` // used in MSG/SAGA
	BinPayloads [][]byte            `json:"-"`
	BranchIDGen `json:"-"`          // used in XA/TCC
	Op          string              `json:"-"` // used in XA/TCC

	QueryPrepared  string `json:"query_prepared,omitempty"` // used in MSG
	Protocol       string `json:"protocol"`
	RollbackReason string `json:"rollback_reason,omitempty" gorm:"-"`
}

TransBase base for all trans

func NewTransBase 

func NewTransBase(gid string, transType string, dtm string, branchID string) *TransBase

NewTransBase new a TransBase

func TransBaseFromQuery 

func TransBaseFromQuery(qs url.Values) *TransBase

TransBaseFromQuery construct transaction info from request

func (*TransBase) WithGlobalTransRequestTimeout 

func (t *TransBase) WithGlobalTransRequestTimeout(timeout int64)

WithGlobalTransRequestTimeout defines global trans request timeout

func (*TransBase) WithRetryLimit 

func (t *TransBase) WithRetryLimit(retryLimit int64)

WithRetryLimit defines global trans retry limit

type TransOptions 

type TransOptions struct {
	WaitResult     bool              `json:"wait_result,omitempty" gorm:"-"`
	TimeoutToFail  int64             `json:"timeout_to_fail,omitempty" gorm:"-"` // for trans type: xa, tcc, unit: second
	RequestTimeout int64             `json:"request_timeout,omitempty" gorm:"-"` // for global trans resets request timeout, unit: second
	RetryInterval  int64             `json:"retry_interval,omitempty" gorm:"-"`  // for trans type: msg saga xa tcc, unit: second
	BranchHeaders  map[string]string `json:"branch_headers,omitempty" gorm:"-"`  // custom branch headers,  dtm server => service api
	Concurrent     bool              `json:"concurrent" gorm:"-"`                // for trans type: saga msg
	RetryLimit     int64             `json:"retry_limit,omitempty" gorm:"-"`     // for trans type: saga
	RetryCount     int64             `json:"retry_count,omitempty" gorm:"-"`     // for trans type: saga
}

TransOptions transaction options

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.