task

Documentation

Overview

Package task is a package for task management

Package task is a package for task management

Package task is a package for task management

Package task is a package for task management

Index

• Constants
• type BrokerConfig
• type CallbackInterface
• type Manager
  • func NewManager() *Manager
  • func (m *Manager) Dispatch(task *types.Task) error
  • func (m *Manager) GetTaskWithID(ctx context.Context, taskid string) (*types.Task, error)
  • func (m *Manager) Init(cfg *ManagerConfig) error
  • func (m *Manager) PatchTaskInfo(ctx context.Context, taskID string, patchs map[string]interface{}) error
  • func (m *Manager) RetryAll(task *types.Task) error
  • func (m *Manager) RetryAt(task *types.Task, stepName string) error
  • func (m *Manager) Run()
  • func (m *Manager) UpdateTask(ctx context.Context, task *types.Task) error
• type ManagerConfig
• type State
  • func NewState(task *types.Task, currentStep string) *State
  • func (s *State) AddCommonParams(key, value string) *State
  • func (s *State) AddStepParams(stepName, key, value string) error
  • func (s *State) GetCommonParams(key string) (string, bool)
  • func (s *State) GetCurrentStep() (*types.Step, bool)
  • func (s *State) GetExtraParams(obj interface{}) error
  • func (s *State) GetStep(stepName string) (*types.Step, bool)
  • func (s *State) GetStepParam(stepName, key string) (string, bool)
  • func (s *State) GetTask() *types.Task
  • func (s *State) SetExtraAll(obj interface{}) error
• type StepWorkerInterface

Constants

const (
	// DefaultWorkerConcurrency default worker concurrency
	DefaultWorkerConcurrency = 10
)

Types

type BrokerConfig

type BrokerConfig struct {
	QueueAddress string `json:"address"`
	Exchange     string `json:"exchange"`
}

BrokerConfig config for go-machinery broker

type CallbackInterface

type CallbackInterface interface {
	GetName() string
	Callback(isSuccess bool, task *types.Task)
}

CallbackInterface that client must implement

type Manager

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

Manager manager for task server

func NewManager

func NewManager() *Manager

NewManager create new manager

func (*Manager) Dispatch

func (m *Manager) Dispatch(task *types.Task) error

Dispatch dispatch task

func (*Manager) GetTaskWithID

func (m *Manager) GetTaskWithID(ctx context.Context, taskid string) (*types.Task, error)

GetTaskWithID get task by taskid

func (*Manager) Init

func (m *Manager) Init(cfg *ManagerConfig) error

Init init machinery server and worker

func (*Manager) PatchTaskInfo

func (m *Manager) PatchTaskInfo(ctx context.Context, taskID string, patchs map[string]interface{}) error

PatchTaskInfo update task info ! warning: modify task status will cause task status not consistent

func (*Manager) RetryAll

func (m *Manager) RetryAll(task *types.Task) error

RetryAll reset status to running and dispatch all tasks

func (*Manager) RetryAt

func (m *Manager) RetryAt(task *types.Task, stepName string) error

RetryAt reset status to running and dispatch tasks which begin with stepName

func (*Manager) Run

func (m *Manager) Run()

Run start worker

func (*Manager) UpdateTask

func (m *Manager) UpdateTask(ctx context.Context, task *types.Task) error

UpdateTask update task ! warning: modify task status will cause task status not consistent

type ManagerConfig

type ManagerConfig struct {
	ModuleName  string
	StepWorkers []StepWorkerInterface
	CallBacks   []CallbackInterface
	WorkerNum   int
	Broker      *BrokerConfig
}

ManagerConfig options for manager

type State

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

State is a struct for task state

func NewState

func NewState(task *types.Task, currentStep string) *State

NewState return state relative to task

func (*State) AddCommonParams

func (s *State) AddCommonParams(key, value string) *State

AddCommonParams add common params

func (*State) AddStepParams

func (s *State) AddStepParams(stepName, key, value string) error

AddStepParams add step params

func (*State) GetCommonParams

func (s *State) GetCommonParams(key string) (string, bool)

GetCommonParams get common params by key

func (*State) GetCurrentStep

func (s *State) GetCurrentStep() (*types.Step, bool)

GetCurrentStep get current step

func (*State) GetExtraParams

func (s *State) GetExtraParams(obj interface{}) error

GetExtraParams get extra params by obj

func (*State) GetStep

func (s *State) GetStep(stepName string) (*types.Step, bool)

GetStep get step by stepName

func (*State) GetStepParam

func (s *State) GetStepParam(stepName, key string) (string, bool)

GetStepParam get step params by key

func (*State) GetTask

func (s *State) GetTask() *types.Task

GetTask get task

func (*State) SetExtraAll

func (s *State) SetExtraAll(obj interface{}) error

SetExtraAll set extra params by obj

type StepWorkerInterface

type StepWorkerInterface interface {
	GetName() string
	DoWork(state *State) error
}

StepWorkerInterface that client must implement