task

Documentation

Index

• func SetTaskConfig(dir string, name string)
• type Batch
• type EtlTask
  • func NewTask(aBatch Batch, syncers ...Syncer) *EtlTask
  • func (e *EtlTask) Execute() error
  • func (e *EtlTask) GetTaskStatus() TaskStatus
  • func (e *EtlTask) IsAvailable() bool
  • func (e *EtlTask) Reset() error
  • func (e *EtlTask) Start() error
  • func (e *EtlTask) StartSyncers()
  • func (e *EtlTask) Stop() error
  • func (e *EtlTask) StopSyncers()
• type LogFunc
• type LogLevel
• type Status
• type Syncer
• type TaskManager
  • func NewTaskManager() *TaskManager
  • func (t *TaskManager) Add(aTask *EtlTask) *TaskManager
  • func (t *TaskManager) Build() error
  • func (t *TaskManager) Execute(taskName string) error
  • func (t *TaskManager) GetAllTasks() map[string]*EtlTask
  • func (t *TaskManager) Reset(taskName string) error
  • func (t *TaskManager) SetLog(log LogFunc)
  • func (t *TaskManager) SetRepo(repo TaskStatusRepository)
  • func (t *TaskManager) Start(taskName string) error
  • func (t *TaskManager) Stop(taskName string) error
• type TaskState
  • func (ts *TaskState) ToString() string
• type TaskStatus
• type TaskStatusRepository
• type TaskType
  • func (tt *TaskType) ToString() string
• type TaskWrapper
  • func NewTaskWrapper(log LogFunc, component *EtlTask) TaskWrapper
  • func (t TaskWrapper) Run()

Functions

func SetTaskConfig

func SetTaskConfig(dir string, name string)

SetTaskConfig set config dir for config file, default search for config, cfgName is no need extension

Types

type Batch

type Batch interface {
	GetName() string
	Begin(t *EtlTask) error
	Reset(t *EtlTask) error
}

Batch interface

type EtlTask

type EtlTask struct {
	Batch            Batch
	Syncers          []Syncer
	Type             TaskType
	PlanTime         string // this time use cron format time
	ResetBeforeBegin bool
	StatusChanged    chan<- *EtlTask
	Status
}

EtlTask indicate a task operate etl

func NewTask

func NewTask(aBatch Batch, syncers ...Syncer) *EtlTask

NewTask create a new etltask with batch

func (*EtlTask) Execute

func (e *EtlTask) Execute() error

Execute a task

func (*EtlTask) GetTaskStatus

func (e *EtlTask) GetTaskStatus() TaskStatus

func (*EtlTask) IsAvailable

func (e *EtlTask) IsAvailable() bool

func (*EtlTask) Reset

func (e *EtlTask) Reset() error

Reset a task

func (*EtlTask) Start

func (e *EtlTask) Start() error

Start a task

func (*EtlTask) StartSyncers

func (e *EtlTask) StartSyncers()

StartSyncers all syncers

func (*EtlTask) Stop

func (e *EtlTask) Stop() error

Stop a task

func (*EtlTask) StopSyncers

func (e *EtlTask) StopSyncers()

StopSyncers all syncers

type LogFunc

type LogFunc func(ll LogLevel, errMsg string)

type LogLevel

type LogLevel string

const (
	Debug LogLevel = "Debug"
	Info  LogLevel = "Info"
	Error LogLevel = "Error"
)

type Status

type Status struct {
	State            TaskState
	LastExecuteTime  time.Time
	LastExecuteState string
	LastExecuteCost  int
}

type Syncer

type Syncer interface {
	Start()
	Stop()
}

Syncer must be singleton

type TaskManager

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

func NewTaskManager

func NewTaskManager() *TaskManager

func (*TaskManager) Add

func (t *TaskManager) Add(aTask *EtlTask) *TaskManager

Add add a name unique task

func (*TaskManager) Build

func (t *TaskManager) Build() error

Build taskmanager, it will do oneshot task first and add plan task to cron then if you specified repo, will recover task status and remove unnecessary task status

func (*TaskManager) Execute

func (t *TaskManager) Execute(taskName string) error

Execute a task

func (*TaskManager) GetAllTasks

func (t *TaskManager) GetAllTasks() map[string]*EtlTask

GetAllTasks this method is commonly used by dashserver

func (*TaskManager) Reset

func (t *TaskManager) Reset(taskName string) error

Reset a task

func (*TaskManager) SetLog

func (t *TaskManager) SetLog(log LogFunc)

SetLog if you want log err messages

func (*TaskManager) SetRepo

func (t *TaskManager) SetRepo(repo TaskStatusRepository)

SetRepo if you need persist task status

func (*TaskManager) Start

func (t *TaskManager) Start(taskName string) error

Start a task

func (*TaskManager) Stop

func (t *TaskManager) Stop(taskName string) error

Stop a task

type TaskState

type TaskState uint

TaskState task state

const (
	// Init task default state
	Init TaskState = iota
	// Executing as it is
	Executing

	// Running as it is
	Running

	// Commplete as it is
	Completed

	// Stopped as it is
	Stopped
)

task state

func (*TaskState) ToString

func (ts *TaskState) ToString() string

ToString transform to enum name

type TaskStatus

type TaskStatus struct {
	TaskName string
	Status
}

type TaskStatusRepository

type TaskStatusRepository interface {
	GetAll() ([]TaskStatus, error)
	InsertOrUpdate(ts TaskStatus) error
	RemoveLegacy(newTs map[string]*EtlTask) error
}

type TaskType

type TaskType uint

TaskType task type

const (
	// OneShot this task will run just one time
	OneShot TaskType = iota
	// Plan when you set this value, you also need to indicate its plan time
	Plan
)

task type

func (*TaskType) ToString

func (tt *TaskType) ToString() string

ToString transform to enum name

type TaskWrapper

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

func NewTaskWrapper

func NewTaskWrapper(log LogFunc, component *EtlTask) TaskWrapper

NewTaskWrapper return a wrapper for cron

func (TaskWrapper) Run

func (t TaskWrapper) Run()

Run it is for cron