task

package

v1.1.2 Latest Latest Go to latest Published: Aug 13, 2020 License: MIT Imports: 6 Imported by: 2

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/ShiningRush/dolphin

Links

Open Source Insights

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()

Constants ¶

This section is empty.

Variables ¶

This section is empty.

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 ¶ added in v1.1.0

func (e *EtlTask) GetTaskStatus() TaskStatus

func (*EtlTask) IsAvailable ¶ added in v1.1.0

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 ¶ added in v1.1.0

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 ¶ added in v1.1.0

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 ¶ added in v1.1.0

type TaskStatus struct {
	TaskName string
	Status
}

type TaskStatusRepository ¶ added in v1.1.0

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

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL