job

Documentation

Overview

Package job provides the job execution and management functionality for the application.

Index

• Constants
• func IsCancelled(ctx context.Context) bool
• type Job
  • func (j *Job) TimeElapsed() time.Duration
• type JobExec
  • func MakeJobExec(fn JobExecFn) JobExec
• type JobExecFn
• type Manager
  • func NewManager() *Manager
  • func (m *Manager) Add(ctx context.Context, description string, e JobExec) int
  • func (m *Manager) CancelAll()
  • func (m *Manager) CancelJob(id int)
  • func (m *Manager) GetJob(id int) *Job
  • func (m *Manager) GetQueue() []Job
  • func (m *Manager) Start(ctx context.Context, description string, e JobExec) int
  • func (m *Manager) Stop()
  • func (m *Manager) Subscribe(ctx context.Context) *ManagerSubscription
• type ManagerSubscription
• type Progress
  • func (p *Progress) AddProcessed(v int)
  • func (p *Progress) AddTotal(total int)
  • func (p *Progress) Definite()
  • func (p *Progress) ExecuteTask(description string, fn func())
  • func (p *Progress) Increment()
  • func (p *Progress) Indefinite()
  • func (p *Progress) SetPercent(percent float64)
  • func (p *Progress) SetProcessed(processed int)
  • func (p *Progress) SetTotal(total int)
• type Status
• type TaskQueue
  • func NewTaskQueue(ctx context.Context, p *Progress, queueSize int, processes int) *TaskQueue
  • func (tq *TaskQueue) Add(description string, fn func(ctx context.Context))
  • func (tq *TaskQueue) Close()

Constants

const ProgressIndefinite float64 = -1

ProgressIndefinite is the special percent value to indicate that the percent progress is not known.

Functions

func IsCancelled

func IsCancelled(ctx context.Context) bool

IsCancelled returns true if cancel has been called on the context.

Types

type Job

type Job struct {
	ID     int
	Status Status
	// details of the current operations of the job
	Details     []string
	Description string
	// Progress in terms of 0 - 1.
	Progress  float64
	StartTime *time.Time
	EndTime   *time.Time
	AddTime   time.Time
	Error     *string
	// contains filtered or unexported fields
}

Job represents the status of a queued or running job.

func (*Job) TimeElapsed

func (j *Job) TimeElapsed() time.Duration

TimeElapsed returns the total time elapsed for the job. If the EndTime is set, then it uses this to calculate the elapsed time, otherwise it uses time.Now.

type JobExec

type JobExec interface {
	Execute(ctx context.Context, progress *Progress) error
}

JobExec represents the implementation of a Job to be executed.

func MakeJobExec

func MakeJobExec(fn JobExecFn) JobExec

MakeJobExec returns a simple JobExec implementation using the provided function.

type JobExecFn

type JobExecFn func(ctx context.Context, progress *Progress) error

type Manager

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

Manager maintains a queue of jobs. Jobs are executed one at a time.

func NewManager

func NewManager() *Manager

NewManager initialises and returns a new Manager.

func (*Manager) Add

func (m *Manager) Add(ctx context.Context, description string, e JobExec) int

Add queues a job.

func (*Manager) CancelAll

func (m *Manager) CancelAll()

CancelAll cancels all of the jobs in the queue. This is the same as calling CancelJob on all jobs in the queue.

func (*Manager) CancelJob

func (m *Manager) CancelJob(id int)

CancelJob cancels the job with the provided id. Jobs that have been started are notified that they are stopping. Jobs that have not yet started are removed from the queue. If no job exists with the provided id, then there is no effect. Likewise, if the job is already cancelled, there is no effect.

func (*Manager) GetJob

func (m *Manager) GetJob(id int) *Job

GetJob returns a copy of the Job for the provided id. Returns nil if the job does not exist.

func (*Manager) GetQueue

func (m *Manager) GetQueue() []Job

GetQueue returns a copy of the current job queue.

func (*Manager) Start

func (m *Manager) Start(ctx context.Context, description string, e JobExec) int

Start adds a job and starts it immediately, concurrently with any other jobs.

func (*Manager) Stop

func (m *Manager) Stop()

Stop is used to stop the dispatcher thread. Once Stop is called, no more Jobs will be processed.

func (*Manager) Subscribe

func (m *Manager) Subscribe(ctx context.Context) *ManagerSubscription

Subscribe subscribes to changes to jobs in the manager queue.

type ManagerSubscription

type ManagerSubscription struct {
	// new jobs are sent to this channel
	NewJob <-chan Job
	// removed jobs are sent to this channel
	RemovedJob <-chan Job
	// updated jobs are sent to this channel
	UpdatedJob <-chan Job
	// contains filtered or unexported fields
}

ManagerSubscription is a collection of channels that will receive updates from the job manager.

type Progress

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

Progress is used by JobExec to communicate updates to the job's progress to the JobManager.

func (*Progress) AddProcessed

func (p *Progress) AddProcessed(v int)

AddProcessed increments the number of processed work units by the provided amount. This is used to calculate the percentage.

func (*Progress) AddTotal

func (p *Progress) AddTotal(total int)

AddTotal adds to the total number of work units. This is used to calculate the progress percentage.

func (*Progress) Definite

func (p *Progress) Definite()

Definite notifies that the total is known.

func (*Progress) ExecuteTask

func (p *Progress) ExecuteTask(description string, fn func())

ExecuteTask executes a task as part of a job. The description is used to populate the Details slice in the parent Job.

func (*Progress) Increment

func (p *Progress) Increment()

Increment increments the number of processed work units. This is used to calculate the percentage. If total is set already, then the number of processed work units will not exceed the total.

func (*Progress) Indefinite

func (p *Progress) Indefinite()

Indefinite sets the progress to an indefinite amount.

func (*Progress) SetPercent

func (p *Progress) SetPercent(percent float64)

SetPercent sets the progress percent directly. This value will be overwritten if Indefinite, SetTotal, Increment or SetProcessed is called. Constrains the percent value between 0 and 1, inclusive.

func (*Progress) SetProcessed

func (p *Progress) SetProcessed(processed int)

SetProcessed sets the number of work units completed. This is used to calculate the progress percentage.

func (*Progress) SetTotal

func (p *Progress) SetTotal(total int)

SetTotal sets the total number of work units and sets definite to true. This is used to calculate the progress percentage.

type Status

type Status string

Status is the status of a Job

const (
	// StatusReady means that the Job is not yet started.
	StatusReady Status = "READY"
	// StatusRunning means that the job is currently running.
	StatusRunning Status = "RUNNING"
	// StatusStopping means that the job is cancelled but is still running.
	StatusStopping Status = "STOPPING"
	// StatusFinished means that the job was completed.
	StatusFinished Status = "FINISHED"
	// StatusCancelled means that the job was cancelled and is now stopped.
	StatusCancelled Status = "CANCELLED"
	// StatusFailed means that the job failed.
	StatusFailed Status = "FAILED"
)

type TaskQueue

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

func NewTaskQueue

func NewTaskQueue(ctx context.Context, p *Progress, queueSize int, processes int) *TaskQueue

func (*TaskQueue) Add

func (tq *TaskQueue) Add(description string, fn func(ctx context.Context))

func (*TaskQueue) Close

func (tq *TaskQueue) Close()