task

Documentation

Index

• Constants
• Variables
• func Register(name string, adapter Instance)
• func RegisterWorker(name string, adapter WorkerInstance)
• type Arg
• type CallbackFunc
• type Config
• type FuncWrap
• type Instance
• type Option
• type Param
• type ParamFromContext
• type Result
• type Task
  • func NewTask(adapterName string, config Config) (adapter Task, err error)
• type Worker
  • func NewWorker(adapterName string, config Config) (adapter Worker, err error)
• type WorkerInstance

Constants

const (
	// TypeAsync - type async
	TypeAsync = "async"
)

const (
	// TypeAsyncWorker - type async worker
	TypeAsyncWorker = "async_worker"
)

Variables

var ErrNotImpl = errors.New("method not impl")

ErrNotImpl - error for not impl

Functions

func Register

func Register(name string, adapter Instance)

Register makes a Task adapter available by the adapter name. If Register is called twice with the same name or if driver is nil, it panics.

func RegisterWorker

func RegisterWorker(name string, adapter WorkerInstance)

RegisterWorker makes a Task worker adapter available by the adapter name. If Register is called twice with the same name or if driver is nil,

Types

type Arg

type Arg struct {
	Name  string      `json:"name"`
	Type  string      `json:"type"`
	Value interface{} `json:"value"`
}

Arg represents a single argument passed to invocation fo a task

type CallbackFunc

type CallbackFunc func(param *Param, err error, message string) error

CallbackFunc - task call back function

type Config

type Config struct {
	// async task
	BrokerType      string `json:"broker_type"`
	Broker          string `json:"broker"`
	DefaultQueue    string `json:"default_queue"`
	BrokerConfig    string `json:"broker_config"`
	ResultBackend   string `json:"result_backend"`
	ResultsExpireIn int    `json:"results_expire_in"`

	FuncWraps map[string]FuncWrap `json:"func_wraps"`
	Logger    log.Logger          `json:"logger"`

	ErrorHandler    func(err error)
	PreTaskHandler  func(param *Param)
	PostTaskHandler func(param *Param)

	// Extend fields
	// Extended fields can be used if there is a special implementation
	Extend1 string `json:"extend_1"`
	Extend2 string `json:"extend_2"`
}

Config - task config

type FuncWrap

type FuncWrap interface {
	// get all task func
	GetTasks() map[string]interface{}
	// stop task by uuid
	StopTask(uuid string) error
}

FuncWrap - task worker func warp

type Instance

type Instance func() Task

Instance is a function create a new Task Instance

type Option

type Option struct {
	ETA          *time.Time `json:"eta"`
	Priority     uint8      `json:"priority"`
	Immutable    bool       `json:"immutable"`
	RetryCount   int        `json:"retry_count"`
	RetryTimeout int        `json:"retry_timeout"`
}

Option task options

type Param

type Param struct {
	UUID        string        `json:"uuid"`
	Name        string        `json:"name"`
	Fun         string        `json:"fun"`
	Option      Option        `json:"option"`
	Args        []Arg         `json:"args"`
	WaitTimeOut time.Duration `json:"wait_time_out"`
}

Param task param

type ParamFromContext

type ParamFromContext func(ctx context.Context) *Param

ParamFromContext - convert context value to Param

var ParamFunc ParamFromContext

ParamFunc - point ParamFromContext impl

type Result

type Result interface {
	// Get - get result with sleep
	Get(sleepDuration time.Duration) ([]reflect.Value, error)

	// GetWithTimeout - get with timeout
	GetWithTimeout(timeoutDuration, sleepDuration time.Duration) ([]reflect.Value, error)
}

Result - task result value

type Task

type Task interface {
	// add new task
	AddTask(param *Param, onSuccess []*Param, onError []*Param, callbacks ...CallbackFunc) error

	// run all task
	// if chain is true, The tasks will be executed in turn, and the return value of the previous
	// task will be used as the parameter of the next task
	Run(chain bool) error

	// stop all task
	Stop() error

	// start gc routine based on config settings.
	StartAndGC(config Config) error
}

Task interface contains all behaviors for Task adapter.

func NewTask

func NewTask(adapterName string, config Config) (adapter Task, err error)

NewTask Create a new Task driver by adapter name and config string. config need to be correct JSON as string: {"server": "localhost:9092", "user": "xxxx", "password":"xxxxx"}. it will start gc automatically.

type Worker

type Worker interface {
	// run worker
	Run() error

	// close worker
	Close() error

	// start gc routine based on config string settings.
	StartAndGC(config Config) error
}

Worker interface contains all behaviors for task worker

func NewWorker

func NewWorker(adapterName string, config Config) (adapter Worker, err error)

NewWorker Create a new Task worker driver by adapter name and config string. config need to be correct JSON as string: it will start gc automatically.

type WorkerInstance

type WorkerInstance func() Worker

WorkerInstance is a function create a new Task worker Instance