task_worker

Documentation

Index

• Constants
• func NewTask(strategy definition.Strategy, task definition.Task, store store.Store, ...) (worker.Worker, error)
• func RegisterTaskInst(task TaskBase)
• func RegisterTaskInstName(name string, task TaskBase)
• func RegisterTaskType(task TaskBase)
• func RegisterTaskTypeName(name string, task TaskBase)
• type BatchExecutor
  • func (m *BatchExecutor) ExecuteOrReturn() bool
• type NormalModel
  • func NewNormalModel(worker *TaskWorker) *NormalModel
  • func (m *NormalModel) LoopOnce()
  • func (m *NormalModel) Stop()
• type SingleExecutor
  • func (m *SingleExecutor) ExecuteOrReturn() bool
• type StreamModel
  • func NewStreamModel(worker *TaskWorker) *StreamModel
  • func (m *StreamModel) LoopOnce()
  • func (m *StreamModel) Stop()
• type TaskBase
• type TaskBatch
• type TaskComparable
• type TaskExecutor
• type TaskModel
• type TaskSingle
• type TaskWorker
  • func (w *TaskWorker) NeedStop() bool
  • func (w *TaskWorker) Start(strategyId, parameter string)
  • func (w *TaskWorker) Stop(strategyId, parameter string)

Constants

const (
	RUNTIME_EMPTY = "<empty>"
)

Functions

func NewTask

func NewTask(strategy definition.Strategy, task definition.Task, store store.Store, schedulerId string) (worker.Worker, error)

NewTask creates a new task and initials necessary fields

Please don't initial TaskWorker manually

func RegisterTaskInst

func RegisterTaskInst(task TaskBase)

RegisterTaskInst registers a task in single instance model with key inferred by its type

func RegisterTaskInstName

func RegisterTaskInstName(name string, task TaskBase)

RegisterTaskInstName registers a task in single instance model with given key

func RegisterTaskType

func RegisterTaskType(task TaskBase)

RegisterTaskType registers a task type with key inferred by its type

func RegisterTaskTypeName

func RegisterTaskTypeName(name string, task TaskBase)

RegisterTaskTypeName registers a task type with key

Types

type BatchExecutor

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

func (*BatchExecutor) ExecuteOrReturn

func (m *BatchExecutor) ExecuteOrReturn() bool

type NormalModel

type NormalModel struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewNormalModel

func NewNormalModel(worker *TaskWorker) *NormalModel

func (*NormalModel) LoopOnce

func (m *NormalModel) LoopOnce()

func (*NormalModel) Stop

func (m *NormalModel) Stop()

type SingleExecutor

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

func (*SingleExecutor) ExecuteOrReturn

func (m *SingleExecutor) ExecuteOrReturn() bool

type StreamModel

type StreamModel struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewStreamModel

func NewStreamModel(worker *TaskWorker) *StreamModel

func (*StreamModel) LoopOnce

func (m *StreamModel) LoopOnce()

func (*StreamModel) Stop

func (m *StreamModel) Stop()

type TaskBase

type TaskBase interface {
	// Select returns tasks to be dealed later.
	//	It will be guaranteed in serial model.
	//	parameter, items, eachFetchNum are from definition of task
	//	ownSign is from name of strategy binded in the form of 'name$ownsign'
	//	It's a kind of relation to strategy but generally task doesn't care about strategy in user's view.
	Select(parameter, ownSign string, items []definition.TaskItem, eachFetchNum int) []interface{}
}

TaskBase defines the task used in scheduling.

type TaskBatch

type TaskBatch interface {
	TaskBase
	// return true if succ false otherwise, but things will still go on
	Execute(tasks []interface{}, ownSign string) bool
}

TaskBatch represents multiple tasks one time(routine) model

type TaskComparable

type TaskComparable interface {
	Less(a, b interface{}) bool
}

type TaskExecutor

type TaskExecutor interface {
	// ExecuteOrReturn returns false indicating no element in queue
	ExecuteOrReturn() bool
}

type TaskModel

type TaskModel interface {
	LoopOnce()
	Stop() // called when worker stopping
}

type TaskSingle

type TaskSingle interface {
	TaskBase
	// return true if succ false otherwise, but things will still go on
	Execute(task interface{}, ownSign string) bool
}

TaskSingle represents one task one time(routine) model

type TaskWorker

type TaskWorker struct {
	sync.Mutex

	// statistics
	NextBeginTime int64
	Statistics    definition.Statistics

	// TimeoutShutdown is the timeout when waiting to close the worker
	TimeoutShutdown time.Duration
	// contains filtered or unexported fields
}

TaskWorker implements a task-driven worker.

Strategy.Bind should be the identifier of task(on console panel).

func (*TaskWorker) NeedStop

func (w *TaskWorker) NeedStop() bool

func (*TaskWorker) Start

func (w *TaskWorker) Start(strategyId, parameter string)

func (*TaskWorker) Stop

func (w *TaskWorker) Stop(strategyId, parameter string)