task

README

异步任务管理模块

Documentation

Index

• Constants
• Variables
• func NewErrorEvent(msg string, taskId string) *event.EventSpec
• func NewInfoEvent(msg string, taskId string) *event.EventSpec
• func StatusCompleteString() []string
• type CONTEXT_TASK_KEY
• type DescribeTaskRequest
  • func NewDescribeTaskRequest(taskId string) *DescribeTaskRequest
• type QueryTaskRequest
  • func NewQueryTaskRequest() *QueryTaskRequest
• type STATUS
  • func (s STATUS) String() string
• type Service
  • func GetService() Service
• type TYPE
• type Task
  • func GetTaskFromCtx(ctx context.Context) *Task
  • func NewTask(spec TaskSpec) *Task
  • func (t *Task) BuildTimeoutCtx() context.Context
  • func (t *Task) Cancel()
  • func (t *Task) Failed(msg string) *Task
  • func (t *Task) Running() *Task
  • func (t *Task) String() string
  • func (t *Task) Success() *Task
  • func (t *Task) TableName() string
• type TaskFunc
• type TaskSpec
  • func NewFnTask(fn TaskFunc, params any) *TaskSpec
  • func (t *TaskSpec) AddWebHook(hs ...*webhook.WebHook) *TaskSpec
  • func (t *TaskSpec) GetFn() TaskFunc
  • func (t *TaskSpec) SetAsync(v bool) *TaskSpec
  • func (t *TaskSpec) SetDescription(desc string) *TaskSpec
  • func (t *TaskSpec) SetLabel(key, value string) *TaskSpec
  • func (t *TaskSpec) SetName(name string) *TaskSpec
• type TaskStatus
  • func NewTaskStatus() *TaskStatus
  • func (s *TaskStatus) SetEndAt(t time.Time)
  • func (s *TaskStatus) SetStartAt(t time.Time)
  • func (s *TaskStatus) SetUpdateAt(t time.Time)

Constants

const (
	APP_NAME = "tasks"
)

const (
	EVENT_LABLE_TASK_ID = "task_id"
)

Variables

var (
	STATUS_MAP = map[STATUS]string{
		STATUS_PENDING:  "PENDDING",
		STATUS_RUNNING:  "RUNNING",
		STATUS_SUCCESS:  "SUCCESS",
		STATUS_FAILED:   "FAILED",
		STATUS_CANCELED: "CANCELED",
	}
	STATUS_COMPLETE = []STATUS{
		STATUS_SUCCESS,
		STATUS_FAILED,
		STATUS_CANCELED,
	}
)

var (
	DEFAULT_TIMEOUT = time.Second * 30
)

Functions

func NewErrorEvent

func NewErrorEvent(msg string, taskId string) *event.EventSpec

func NewInfoEvent

func NewInfoEvent(msg string, taskId string) *event.EventSpec

func StatusCompleteString

func StatusCompleteString() []string

Types

type CONTEXT_TASK_KEY

type CONTEXT_TASK_KEY struct{}

type DescribeTaskRequest

type DescribeTaskRequest struct {
	TaskId string `json:"task_id"`
}

func NewDescribeTaskRequest

func NewDescribeTaskRequest(taskId string) *DescribeTaskRequest

type QueryTaskRequest

type QueryTaskRequest struct {
	request.PageRequest
}

func NewQueryTaskRequest

func NewQueryTaskRequest() *QueryTaskRequest

type STATUS

type STATUS int

const (
	// 任务处于挂起, 队列中
	STATUS_PENDING STATUS = iota
	// 任务正在运行
	STATUS_RUNNING
	// 任务已完成
	STATUS_SUCCESS
	// 任务失败
	STATUS_FAILED
	// 任务已取消
	STATUS_CANCELED
)

func (STATUS) String

func (s STATUS) String() string

type Service

type Service interface {
	// 任务执行
	Run(context.Context, *TaskSpec) *Task
	// 查询任务列表
	QueryTask(context.Context, *QueryTaskRequest) (*types.Set[*Task], error)
	// 查询任务详情
	DescribeTask(context.Context, *DescribeTaskRequest) (*Task, error)
}

func GetService

func GetService() Service

type TYPE

type TYPE string

const (
	// 任务是一个函数
	TYPE_FUNCTION TYPE = "function"
)

type Task

type Task struct {
	// 任务Id
	Id string `json:"id" gorm:"column:id;type:string;primary_key;" unique:"true" description:"Id"`
	// 创建时间
	CreatedAt time.Time `` /* 126-byte string literal not displayed */
	// 任务定义
	TaskSpec
	// 任务状态
	TaskStatus
}

func GetTaskFromCtx

func GetTaskFromCtx(ctx context.Context) *Task

func NewTask

func NewTask(spec TaskSpec) *Task

func (*Task) BuildTimeoutCtx

func (t *Task) BuildTimeoutCtx() context.Context

注入上下文当中

func (*Task) Cancel

func (t *Task) Cancel()

func (*Task) Failed

func (t *Task) Failed(msg string) *Task

func (*Task) Running

func (t *Task) Running() *Task

func (*Task) String

func (t *Task) String() string

func (*Task) Success

func (t *Task) Success() *Task

func (*Task) TableName

func (t *Task) TableName() string

type TaskFunc

type TaskFunc func(ctx context.Context, req any) error

type TaskSpec

type TaskSpec struct {
	// 是否异步执行
	Async bool `json:"async" gorm:"column:async;" description:"是否异步执行"`
	// 异步执行时的超时时间
	Timeout string `json:"timeout" gorm:"column:timeout;" description:"异步执行时的超时时间"`
	// 任务类型
	Type TYPE `json:"type" gorm:"column:id;type:varchar(60);" description:"任务类型"`
	// 任务名称
	Name string `json:"name" gorm:"column:name;type:varchar(200);" description:"任务名称"`
	// 任务名称
	Description string `json:"description" gorm:"column:description;type:text;" description:"任务描述"`
	// 尝试执行,用于做执行前检查
	DryRun bool `json:"dryrun" gorm:"column:dryrun;type:bool;" description:"尝试执行,用于做执行前检查"`
	// 任务的参数
	Params any `json:"params" gorm:"column:params;serializer:json;type:json" description:"任务参数"`
	// 任务标签
	Label map[string]string `json:"label" bson:"label" gorm:"column:label;serializer:json;type:json" description:"任务标签" optional:"true"`

	// 任务执行结束回调
	WebHooks []*webhook.WebHook `` /* 138-byte string literal not displayed */
	// contains filtered or unexported fields
}

func NewFnTask

func NewFnTask(fn TaskFunc, params any) *TaskSpec

func (*TaskSpec) AddWebHook

func (t *TaskSpec) AddWebHook(hs ...*webhook.WebHook) *TaskSpec

func (*TaskSpec) GetFn

func (t *TaskSpec) GetFn() TaskFunc

func (*TaskSpec) SetAsync

func (t *TaskSpec) SetAsync(v bool) *TaskSpec

func (*TaskSpec) SetDescription

func (t *TaskSpec) SetDescription(desc string) *TaskSpec

func (*TaskSpec) SetLabel

func (t *TaskSpec) SetLabel(key, value string) *TaskSpec

func (*TaskSpec) SetName

func (t *TaskSpec) SetName(name string) *TaskSpec

type TaskStatus

type TaskStatus struct {
	// 开始执行时间
	StartAt *time.Time `json:"start_at" gorm:"column:start_at;type:timestamp;" description:"开始执行时间"`
	// 执行结束的时间
	EndAt *time.Time `json:"end_at" gorm:"column:end_at;type:timestamp;" description:"执行结束的时间"`
	// 任务状态更新时间
	UpdateAt *time.Time `json:"update_at" gorm:"column:update_at;type:timestamp;" description:"任务状态更新时间"`
	// 任务执行状态
	Status STATUS `json:"status" gorm:"column:status;type:tinyint(2);" description:"任务执行状态"`
	// 失败信息
	Message string `json:"message" gorm:"column:message;type:text;" description:"失败信息"`

	// 执行过程中的事件, 执行日志
	Events []*event.Event `json:"events" gorm:"column:events;type:json;serializer:json;" description:"执行过程中的事件"`
	// contains filtered or unexported fields
}

func NewTaskStatus

func NewTaskStatus() *TaskStatus

func (*TaskStatus) SetEndAt

func (s *TaskStatus) SetEndAt(t time.Time)

func (*TaskStatus) SetStartAt

func (s *TaskStatus) SetStartAt(t time.Time)

func (*TaskStatus) SetUpdateAt

func (s *TaskStatus) SetUpdateAt(t time.Time)