Documentation
¶
Index ¶
- type Task
- type TaskRunner
- func (r *TaskRunner) CancelTask(ctx context.Context, id task.ID) error
- func (r *TaskRunner) GetTask(ctx context.Context, id task.ID) (task.Task, error)
- func (r *TaskRunner) GetTaskState(ctx context.Context, id task.ID) (*task.State, error)
- func (r *TaskRunner) ListTasks(ctx context.Context) ([]task.StateHeader, error)
- func (r *TaskRunner) RegisterFactory(taskType task.Type, factory task.Factory)
- func (r *TaskRunner) RegisterTask(taskType task.Type, handler task.Handler)
- func (r *TaskRunner) Run(ctx context.Context) error
- func (r *TaskRunner) ScheduleTask(ctx context.Context, tsk task.Task) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Task ¶
type Task struct {
ID string `gorm:"primaryKey;autoIncrement:false"`
Type string `gorm:"index"`
Payload []byte
Status string `gorm:"index"`
ScheduledAt time.Time
FinishedAt *time.Time
Progress float32
Message string
Error string
CreatedAt time.Time
UpdatedAt time.Time
}
Task is the persisted representation of a task.Task: its serialized payload (produced by the task's json.Marshaler) plus the mutable execution state. It is the single source of truth for the persistent runner, allowing pending and interrupted tasks to be resumed after a restart.
type TaskRunner ¶
type TaskRunner struct {
// contains filtered or unexported fields
}
TaskRunner is a persistent task.Runner backed by a gorm database. Scheduled tasks are serialized (through their json.Marshaler) and stored, so pending tasks — and tasks that were still running when the process stopped — are resumed on the next Run. Deserialization relies on factories registered with RegisterFactory (task.PersistentRunner).
func NewTaskRunner ¶
func NewTaskRunner(db *gorm.DB, parallelism int, cleanupDelay time.Duration, cleanupInterval time.Duration) *TaskRunner
NewTaskRunner creates a persistent runner backed by db. The schema is migrated lazily on the first database access. Register the task handlers with RegisterTask and the deserialization factories with RegisterFactory before calling Run so pending tasks can be resumed.
func NewTaskRunnerWithQueue ¶
func NewTaskRunnerWithQueue(db *gorm.DB, parallelism int, queueSize int, errOnFull bool, cleanupDelay time.Duration, cleanupInterval time.Duration) *TaskRunner
NewTaskRunnerWithQueue is NewTaskRunner with control over the in-memory queue size and whether ScheduleTask fails (ErrQueueFull) when the queue is full.
func (*TaskRunner) CancelTask ¶
CancelTask implements task.Runner.
func (*TaskRunner) GetTask ¶
GetTask implements task.Runner. It rebuilds the concrete task from its persisted payload using the registered factory.
func (*TaskRunner) GetTaskState ¶
GetTaskState implements task.Runner.
func (*TaskRunner) ListTasks ¶
func (r *TaskRunner) ListTasks(ctx context.Context) ([]task.StateHeader, error)
ListTasks implements task.Runner.
func (*TaskRunner) RegisterFactory ¶
func (r *TaskRunner) RegisterFactory(taskType task.Type, factory task.Factory)
RegisterFactory implements task.PersistentRunner. The factory rebuilds a concrete task from its persisted payload so it can be resumed or fetched.
func (*TaskRunner) RegisterTask ¶
func (r *TaskRunner) RegisterTask(taskType task.Type, handler task.Handler)
RegisterTask implements task.Runner.
func (*TaskRunner) Run ¶
func (r *TaskRunner) Run(ctx context.Context) error
Run implements task.Runner. It first resumes persisted work (pending tasks and tasks left running by a previous, interrupted process), then serves the queue with a fixed worker pool until the context is canceled. On shutdown it lets in-flight tasks drain before returning.
func (*TaskRunner) ScheduleTask ¶
ScheduleTask implements task.Runner. The task is persisted as pending before being enqueued, so it survives a restart even if it is never picked up by a worker in this process.
Source Files