importer

Documentation

Index

• func CreateConfigFile(taskdir string, cfgBytes []byte) (string, error)
• func CreateNewTaskDir(rootDir string, id string) (string, error)
• func DeleteImportTask(tasksDir, taskID, address, username string) error
• func GetImportTask(taskID, address, username string) (*types.GetImportTaskData, error)
• func GetManyImportTask(address, username, space string, pageIndex, pageSize int) (*types.GetManyImportTaskData, error)
• func InitTaskStatus()
• func StartImport(taskID string) (err error)
• func StopImportTask(taskID, address, username string) error
• type Client
• type ImportResult
• type Task
  • func (t *Task) UpdateQueryStats() error
• type TaskDb
  • func (t *TaskDb) DelTaskEffect(ID string) error
  • func (t *TaskDb) DelTaskInfo(ID string) error
  • func (t *TaskDb) FindTaskInfoByAddressAndUser(address, user, space string, pageIndex, pageSize int) ([]*db.TaskInfo, int64, error)
  • func (t *TaskDb) FindTaskInfoByIdAndAddresssAndUser(id, address, user string) (*db.TaskInfo, error)
  • func (t *TaskDb) InsertTaskEffect(taskEffect *db.TaskEffect) error
  • func (t *TaskDb) InsertTaskInfo(info *db.TaskInfo) error
  • func (t *TaskDb) UpdateProcessingTasks2Aborted() error
  • func (t *TaskDb) UpdateTaskEffect(taskEffect *db.TaskEffect) error
  • func (t *TaskDb) UpdateTaskInfo(info *db.TaskInfo) error
• type TaskMgr
  • func GetTaskMgr() *TaskMgr
  • func (mgr *TaskMgr) AbortTask(taskID string) (err error)
  • func (mgr *TaskMgr) DelTask(tasksDir, taskID string) error
  • func (mgr *TaskMgr) FinishTask(taskID string) (err error)
  • func (mgr *TaskMgr) GetTask(taskID string) (*Task, bool)
  • func (mgr *TaskMgr) NewTask(id, host, user, taskName, rawCfg string, cfg importconfig.Configurator) (*Task, error)
  • func (mgr *TaskMgr) NewTaskDraft(id, host, user, taskName, space, rawCfg string) error
  • func (mgr *TaskMgr) NewTaskEffect(taskEffect *db.TaskEffect) error
  • func (mgr *TaskMgr) PutTask(taskID string, task *Task)
  • func (mgr *TaskMgr) StopTask(taskID string) error
  • func (mgr *TaskMgr) StorePartTaskLog(taskID string) error
  • func (mgr *TaskMgr) TurnDraftToTask(id, taskName, rawCfg string, cfg importconfig.Configurator) (*Task, error)
  • func (mgr *TaskMgr) UpdateTaskDraft(id, taskName, space, rawCfg string) error
  • func (mgr *TaskMgr) UpdateTaskInfo(taskID string) error
• type TaskStatus
  • func NewTaskStatus(status string) TaskStatus
  • func (status TaskStatus) String() string

Functions

func CreateConfigFile

func CreateConfigFile(taskdir string, cfgBytes []byte) (string, error)

func CreateNewTaskDir

func CreateNewTaskDir(rootDir string, id string) (string, error)

func DeleteImportTask

func DeleteImportTask(tasksDir, taskID, address, username string) error

func GetImportTask

func GetImportTask(taskID, address, username string) (*types.GetImportTaskData, error)

func GetManyImportTask

func GetManyImportTask(address, username, space string, pageIndex, pageSize int) (*types.GetManyImportTaskData, error)

func InitTaskStatus

func InitTaskStatus()

func StartImport

func StartImport(taskID string) (err error)

func StopImportTask

func StopImportTask(taskID, address, username string) error

Types

type Client

type Client struct {
	Cfg        config.Configurator `json:"cfg,omitempty"`
	Logger     logger.Logger       `json:"logger,omitempty"`
	Manager    manager.Manager     `json:"manager,omitempty"`
	HasStarted bool                `json:"has_started,omitempty"`
}

type ImportResult

type ImportResult struct {
	TaskId      string `json:"taskId"`
	TimeCost    string `json:"timeCost"` // Milliseconds
	FailedRows  int64  `json:"failedRows"`
	ErrorResult struct {
		ErrorCode int    `json:"errorCode"`
		ErrorMsg  string `json:"errorMsg"`
	}
}

type Task

type Task struct {
	Client   *Client      `json:"client,omitempty"`
	TaskInfo *db.TaskInfo `json:"task_info,omitempty"`
}

func (*Task) UpdateQueryStats

func (t *Task) UpdateQueryStats() error

type TaskDb

type TaskDb struct {
	*gorm.DB
}

func (*TaskDb) DelTaskEffect

func (t *TaskDb) DelTaskEffect(ID string) error

func (*TaskDb) DelTaskInfo

func (t *TaskDb) DelTaskInfo(ID string) error

func (*TaskDb) FindTaskInfoByAddressAndUser

func (t *TaskDb) FindTaskInfoByAddressAndUser(address, user, space string, pageIndex, pageSize int) ([]*db.TaskInfo, int64, error)

func (*TaskDb) FindTaskInfoByIdAndAddresssAndUser

func (t *TaskDb) FindTaskInfoByIdAndAddresssAndUser(id, address, user string) (*db.TaskInfo, error)

FindTaskInfoByIdAndAddresssAndUser used to check whether the task belongs to the user

func (*TaskDb) InsertTaskEffect

func (t *TaskDb) InsertTaskEffect(taskEffect *db.TaskEffect) error

func (*TaskDb) InsertTaskInfo

func (t *TaskDb) InsertTaskInfo(info *db.TaskInfo) error

func (*TaskDb) UpdateProcessingTasks2Aborted

func (t *TaskDb) UpdateProcessingTasks2Aborted() error

func (*TaskDb) UpdateTaskEffect

func (t *TaskDb) UpdateTaskEffect(taskEffect *db.TaskEffect) error

func (*TaskDb) UpdateTaskInfo

func (t *TaskDb) UpdateTaskInfo(info *db.TaskInfo) error

type TaskMgr

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

func GetTaskMgr

func GetTaskMgr() *TaskMgr

func (*TaskMgr) AbortTask

func (mgr *TaskMgr) AbortTask(taskID string) (err error)

func (*TaskMgr) DelTask

func (mgr *TaskMgr) DelTask(tasksDir, taskID string) error

func (*TaskMgr) FinishTask

func (mgr *TaskMgr) FinishTask(taskID string) (err error)

FinishTask will query task stats

• delete task in the map
• update taskInfo in db
• update taskEffect in db

func (*TaskMgr) GetTask

func (mgr *TaskMgr) GetTask(taskID string) (*Task, bool)

GetTask get task from map and local sql

func (*TaskMgr) NewTask

func (mgr *TaskMgr) NewTask(id, host, user, taskName, rawCfg string, cfg importconfig.Configurator) (*Task, error)

func (*TaskMgr) NewTaskDraft

func (mgr *TaskMgr) NewTaskDraft(id, host, user, taskName, space, rawCfg string) error

func (*TaskMgr) NewTaskEffect

func (mgr *TaskMgr) NewTaskEffect(taskEffect *db.TaskEffect) error

func (*TaskMgr) PutTask

func (mgr *TaskMgr) PutTask(taskID string, task *Task)

PutTask put task into tasks map

func (*TaskMgr) StopTask

func (mgr *TaskMgr) StopTask(taskID string) error

StopTask will change the task status to `Stoped`, and then call FinishTask

func (*TaskMgr) StorePartTaskLog

func (mgr *TaskMgr) StorePartTaskLog(taskID string) error

func (*TaskMgr) TurnDraftToTask

func (mgr *TaskMgr) TurnDraftToTask(id, taskName, rawCfg string, cfg importconfig.Configurator) (*Task, error)

func (*TaskMgr) UpdateTaskDraft

func (mgr *TaskMgr) UpdateTaskDraft(id, taskName, space, rawCfg string) error

func (*TaskMgr) UpdateTaskInfo

func (mgr *TaskMgr) UpdateTaskInfo(taskID string) error

UpdateTaskInfo will query task stats, update task in the map and update the taskInfo in local sql

type TaskStatus

type TaskStatus int

const (
	StatusUnknown TaskStatus = iota
	Finished
	Stoped
	Processing
	NotExisted
	Aborted
	Draft
)

the task in memory (map) has 2 status: processing, aborted; and the task in local sql has 2 status: finished, stoped;

func NewTaskStatus

func NewTaskStatus(status string) TaskStatus

func (TaskStatus) String

func (status TaskStatus) String() string