compense

README

compense

compense

Documentation

Index

• func ByDb(core *xorm.Engine, opts ...Option)
• type CompensationOption
  • func NewOption(opts ...Option) *CompensationOption
• type DBCompensator
  • func (r *DBCompensator) BatchRetry(ctx context.Context, group string, limit int32) error
  • func (r *DBCompensator) Execute(ctx context.Context, task *Task) (int64, IOutTasker, error)
  • func (r *DBCompensator) ExecuteAsync(ctx context.Context, task *Task) (int64, IOutTasker, error)
  • func (r *DBCompensator) Logger() ILogger
  • func (r *DBCompensator) Register(key string, exec Execute) error
  • func (r *DBCompensator) TaskByID(ctx context.Context, id string) (IOutTasker, error)
• type Execute
• type ICompensator
  • func Global() ICompensator
  • func NewDBCompensator(core *xorm.Engine, option *CompensationOption) ICompensator
  • func NewDBCompensatorWithOption(core *xorm.Engine, options ...Option) ICompensator
• type IExecutor
• type ILogger
• type IOutTasker
• type ITasker
• type Option
  • func WithExecCallBack(v func(string)) Option
  • func WithFuncMap(v map[string]Execute) Option
  • func WithLogger(v ILogger) Option
  • func WithNewAsyncCtx(v func(ctx context.Context) context.Context) Option
  • func WithOffset(v time.Duration) Option
• type Task
  • func (p *Task) GetArgs() string
  • func (p *Task) GetDeadline() int64
  • func (p *Task) GetGroup() string
  • func (p *Task) GetID() interface{}
  • func (p *Task) GetLastError() string
  • func (p *Task) GetName() string
  • func (p *Task) GetOwner() string
  • func (p *Task) GetParams() (map[string]interface{}, error)
  • func (p *Task) GetRetries() int32
  • func (p *Task) GetTimeout() time.Duration
  • func (p *Task) TableName() string

Functions

func ByDb

func ByDb(core *xorm.Engine, opts ...Option)

Types

type CompensationOption

type CompensationOption struct {
	// contains filtered or unexported fields
}

CompensationOption compensation option

func NewOption

func NewOption(opts ...Option) *CompensationOption

NewOption *

• @Description:
• @param opts
• @return CompensationOption

type DBCompensator

type DBCompensator struct {
	// contains filtered or unexported fields
}

DBCompensator use db compensate

func (*DBCompensator) BatchRetry

func (r *DBCompensator) BatchRetry(ctx context.Context, group string, limit int32) error

BatchRetry batch retry func

func (*DBCompensator) Execute

func (r *DBCompensator) Execute(ctx context.Context, task *Task) (int64, IOutTasker, error)

Execute exec

func (*DBCompensator) ExecuteAsync

func (r *DBCompensator) ExecuteAsync(ctx context.Context, task *Task) (int64, IOutTasker, error)

ExecuteAsync exec async

func (*DBCompensator) Logger

func (r *DBCompensator) Logger() ILogger

Logger use log

func (*DBCompensator) Register

func (r *DBCompensator) Register(key string, exec Execute) error

Register register execute func

func (*DBCompensator) TaskByID

func (r *DBCompensator) TaskByID(ctx context.Context, id string) (IOutTasker, error)

TaskByID only read record by id without locker

type Execute

type Execute func(ctx context.Context, tasker ITasker) error

type ICompensator

type ICompensator interface {
	Register(name string, exec Execute) error
	Execute(ctx context.Context, task *Task) (int64, IOutTasker, error)
	ExecuteAsync(ctx context.Context, task *Task) (int64, IOutTasker, error)
	BatchRetry(ctx context.Context, group string, limit int32) error
	TaskByID(ctx context.Context, id string) (IOutTasker, error)
}

func Global

func Global() ICompensator

func NewDBCompensator

func NewDBCompensator(core *xorm.Engine, option *CompensationOption) ICompensator

NewDBCompensator new a compensator base on db

func NewDBCompensatorWithOption

func NewDBCompensatorWithOption(core *xorm.Engine, options ...Option) ICompensator

NewDBCompensatorWithOption new a compensator base on db

type IExecutor

type IExecutor interface {
	Registry(name string, exec Execute)
}

type ILogger

type ILogger interface {
	DebugCtxf(ctx context.Context, format string, args ...interface{})
	InfoCtxf(ctx context.Context, format string, args ...interface{})
	WarnCtxf(ctx context.Context, format string, args ...interface{})
	ErrorCtxf(ctx context.Context, format string, args ...interface{})
}

type IOutTasker

type IOutTasker interface {
	ITasker
	GetLastError() string
	GetRetries() int32
}

type ITasker

type ITasker interface {
	GetID() interface{}
	GetGroup() string
	GetName() string
	GetArgs() string
	GetParams() (map[string]interface{}, error)
	GetDeadline() int64
	GetOwner() string
}

type Option

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

func WithExecCallBack

func WithExecCallBack(v func(string)) Option

WithExecCallBack *

• @Description:
• @param v
• @return Option

func WithFuncMap

func WithFuncMap(v map[string]Execute) Option

WithFuncMap *

• @Description:
• @param v
• @return Option

func WithLogger

func WithLogger(v ILogger) Option

WithLogger *

• @Description:
• @param v
• @return Option

func WithNewAsyncCtx

func WithNewAsyncCtx(v func(ctx context.Context) context.Context) Option

WithNewAsyncCtx *

• @Description:
• @param v
• @return Option

func WithOffset

func WithOffset(v time.Duration) Option

WithOffset *

• @Description: suggest 180 seconds
• @param v
• @return Option

type Task

type Task struct {
	ID           string      `json:"id" xorm:"varchar(36) pk comment('唯一主键') 'id'"`                 //  样例
	TimeStamp    int64       `json:"timeStamp" xorm:"bigint comment('消息时间戳') 'time_stamp'"`         //  样例
	Expiration   int64       `json:"expiration" xorm:"bigint comment('生效时间戳') 'expiration'"`        //  样例
	NotBefore    int64       `json:"notBefore" xorm:"bigint comment('生效时间戳') 'not_before'"`         //  样例
	Timeout      int64       `json:"timeout" xorm:"bigint comment('执行超时ms') 'timeout'"`             //  样例
	AppID        string      `json:"appId" xorm:"varchar(32) comment('消息源AppId') 'app_id'"`         //  样例
	Group        string      `json:"group" xorm:"varchar(32) comment('消息分组') 'group'"`              //  样例
	Name         string      `json:"name" xorm:"varchar(32) comment('消息名') 'name'"`                 //  样例
	Body         string      `json:"body" xorm:"text comment('消息主体') 'body'"`                       //  样例
	Retries      int32       `json:"retries" xorm:"int comment('重试次数') 'retries'"`                  //  样例
	LockExpire   int64       `json:"lockExpire" xorm:"bigint comment('锁过期时间') 'lock_expire'"`       //  样例
	Source       string      `json:"source" xorm:"varchar(36) comment('来源') 'source'"`              //  样例
	Locker       string      `json:"locker" xorm:"varchar(36) comment('执行者') 'locker'"`             //  样例
	LastError    string      `json:"lastError" xorm:"varchar(255) comment('最后失败原因') 'last_error'"`  //  样例
	LastExecTime int64       `json:"lastExecTime" xorm:"bigint comment('最后执行时间') 'last_exec_time'"` //  样例
	Raw          interface{} `json:"-" xorm:"-"`                                                    //  样例
}

func (*Task) GetArgs

func (p *Task) GetArgs() string

func (*Task) GetDeadline

func (p *Task) GetDeadline() int64

func (*Task) GetGroup

func (p *Task) GetGroup() string

func (*Task) GetID

func (p *Task) GetID() interface{}

func (*Task) GetLastError

func (p *Task) GetLastError() string

func (*Task) GetName

func (p *Task) GetName() string

func (*Task) GetOwner

func (p *Task) GetOwner() string

func (*Task) GetParams

func (p *Task) GetParams() (map[string]interface{}, error)

func (*Task) GetRetries

func (p *Task) GetRetries() int32

func (*Task) GetTimeout

func (p *Task) GetTimeout() time.Duration

func (*Task) TableName

func (p *Task) TableName() string