tasks

Documentation

Index

• Variables
• type Runner
  • func NewRunner(reg kit.Registry, b db.Backend, model kit.Model) *Runner
  • func (r *Runner) Backend() db.Backend
  • func (r *Runner) GetTaskCheckInterval() time.Duration
  • func (r *Runner) GetTaskSpecs() map[string]kit.TaskSpec
  • func (r *Runner) MaximumConcurrentTasks() int
  • func (r *Runner) RegisterTask(spec kit.TaskSpec)
  • func (r *Runner) Registry() kit.Registry
  • func (r *Runner) Run() apperror.Error
  • func (r *Runner) SetBackend(backend db.Backend)
  • func (r *Runner) SetMaximumConcurrentTasks(count int)
  • func (r *Runner) SetRegistry(registry kit.Registry)
  • func (r *Runner) SetTaskCheckInterval(duration time.Duration)
  • func (r *Runner) Shutdown() chan bool
• type Service
  • func NewService(reg kit.Registry, b db.Backend) *Service
  • func (s *Service) GetTask(id string) (kit.Task, apperror.Error)
  • func (s *Service) Queue(task kit.Task) apperror.Error
• type Task
  • func (Task) Collection() string
  • func (t *Task) GetCreatedAt() time.Time
  • func (t Task) GetData() interface{}
  • func (t Task) GetError() string
  • func (t Task) GetFinishedAt() *time.Time
  • func (t Task) GetLog() string
  • func (t Task) GetName() string
  • func (t *Task) GetPriority() int
  • func (t *Task) GetProgress() int
  • func (t Task) GetResult() interface{}
  • func (t *Task) GetRunAt() *time.Time
  • func (t Task) GetStartedAt() *time.Time
  • func (t Task) GetTryCount() int
  • func (t *Task) IsCancelled() bool
  • func (t Task) IsComplete() bool
  • func (t *Task) IsRunning() bool
  • func (t Task) IsSuccess() bool
  • func (t *Task) SetCreatedAt(tm time.Time)
  • func (t *Task) SetData(data interface{})
  • func (t *Task) SetError(err string)
  • func (t *Task) SetFinishedAt(tm *time.Time)
  • func (t *Task) SetIsCancelled(x bool)
  • func (t *Task) SetIsComplete(flag bool)
  • func (t *Task) SetIsRunning(x bool)
  • func (t *Task) SetIsSuccess(flag bool)
  • func (t *Task) SetLog(log string)
  • func (t *Task) SetName(name string)
  • func (t *Task) SetPriority(x int)
  • func (t *Task) SetProgress(x int)
  • func (t *Task) SetResult(result interface{})
  • func (t *Task) SetRunAt(x *time.Time)
  • func (t *Task) SetStartedAt(tm *time.Time)
  • func (t *Task) SetTryCount(count int)
• type TaskIntId
  • func (t TaskIntId) GetUserId() interface{}
  • func (t TaskIntId) SetUserId(id interface{})
• type TaskSpec
  • func (s TaskSpec) GetAllowedRetries() int
  • func (s TaskSpec) GetHandler() kit.TaskHandler
  • func (s TaskSpec) GetName() string
  • func (s TaskSpec) GetOnCompleteHandler() kit.TaskOnCompleteHandler
  • func (s TaskSpec) GetRetryInterval() time.Duration
• type TaskStrId
  • func (t TaskStrId) GetUserId() interface{}
  • func (t TaskStrId) SetUserId(id interface{})

Variables

var CancelTaskMethod = &methods.Method{
	Name:     "task.cancel",
	Blocking: false,
	Handler: func(registry kit.Registry, r kit.Request, unblock func()) kit.Response {
		user := r.GetUser()
		if user == nil {
			return kit.NewErrorResponse("not_authenticated", true)
		}

		taskId := r.GetData()
		if taskId == nil {
			return kit.NewErrorResponse("invalid_task_id", "Expected 'data' to be the task Id.", true)
		}

		backend := registry.DefaultBackend()
		rawTask, err := backend.FindOne("tasks", taskId)
		if err != nil {
			return kit.NewErrorResponse(err)
		} else if rawTask == nil {
			return kit.NewErrorResponse("not_found", "Task does not exist.")
		}

		task := rawTask.(kit.Task)

		if !(user.HasRole("admin") || user.GetId() == task.GetUserId()) {
			return kit.NewErrorResponse("permission_denied")
		}

		if task.IsComplete() {
			return kit.NewErrorResponse("task_complete", "Can't cancel a completed task.")
		} else if task.IsRunning() {
			return kit.NewErrorResponse("task_running", "Can't cancel a running task.")
		}

		if err := backend.Update(task); err != nil {
			return kit.NewErrorResponse(err)
		}

		return &kit.AppResponse{
			Data: map[string]interface{}{"success": true},
		}
	},
}

var RetryTaskMethod = &methods.Method{
	Name:     "task.retry",
	Blocking: false,
	Handler: func(registry kit.Registry, r kit.Request, unblock func()) kit.Response {
		user := r.GetUser()
		if user == nil {
			return kit.NewErrorResponse("not_authenticated", true)
		}

		taskId := r.GetData()
		if taskId == nil {
			return kit.NewErrorResponse("invalid_task_id", "Expected 'data' to be the task Id.", true)
		}

		backend := registry.DefaultBackend()
		rawTask, err := backend.FindOne("tasks", taskId)
		if err != nil {
			return kit.NewErrorResponse(err)
		} else if rawTask == nil {
			return kit.NewErrorResponse("not_found", "Task does not exist.")
		}

		task := rawTask.(kit.Task)

		if !(user.HasRole("admin") || user.GetId() == task.GetUserId()) {
			return kit.NewErrorResponse("permission_denied")
		}

		if task.IsSuccess() {
			return kit.NewErrorResponse("task_succeeded", "Can't retry a succeeded task")
		} else if !task.IsComplete() {
			return kit.NewErrorResponse("task_not_complete", "Can't retry a task that has not completed yet.")
		}

		task.SetIsComplete(false)
		task.SetTryCount(0)
		task.SetRunAt(nil)

		if err := backend.Update(task); err != nil {
			return kit.NewErrorResponse(err)
		}

		return &kit.AppResponse{
			Data: map[string]interface{}{"success": true},
		}
	},
}

Types

type Runner

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

func NewRunner

func NewRunner(reg kit.Registry, b db.Backend, model kit.Model) *Runner

func (*Runner) Backend

func (r *Runner) Backend() db.Backend

func (*Runner) GetTaskCheckInterval

func (r *Runner) GetTaskCheckInterval() time.Duration

func (*Runner) GetTaskSpecs

func (r *Runner) GetTaskSpecs() map[string]kit.TaskSpec

GetTaskSpecs returns a slice with all registered tasks.

func (*Runner) MaximumConcurrentTasks

func (r *Runner) MaximumConcurrentTasks() int

func (*Runner) RegisterTask

func (r *Runner) RegisterTask(spec kit.TaskSpec)

func (*Runner) Registry

func (r *Runner) Registry() kit.Registry

func (*Runner) Run

func (r *Runner) Run() apperror.Error

func (*Runner) SetBackend

func (r *Runner) SetBackend(backend db.Backend)

func (*Runner) SetMaximumConcurrentTasks

func (r *Runner) SetMaximumConcurrentTasks(count int)

func (*Runner) SetRegistry

func (r *Runner) SetRegistry(registry kit.Registry)

func (*Runner) SetTaskCheckInterval

func (r *Runner) SetTaskCheckInterval(duration time.Duration)

func (*Runner) Shutdown

func (r *Runner) Shutdown() chan bool

type Service

type Service struct {
	Runner
}

func NewService

func NewService(reg kit.Registry, b db.Backend) *Service

func (*Service) GetTask

func (s *Service) GetTask(id string) (kit.Task, apperror.Error)

func (*Service) Queue

func (s *Service) Queue(task kit.Task) apperror.Error

type Task

type Task struct {
	Name     string      `db:"required"`
	Data     interface{} `db:"marshal"`
	RunAt    *time.Time
	Priority int
	Progress int

	CreatedAt time.Time

	Cancelled bool

	Result interface{} `db:"marshal"`

	TryCount int

	StartedAt  *time.Time
	FinishedAt *time.Time

	Running  bool
	Complete bool
	Success  bool

	Error string
	Log   string
}

func (Task) Collection

func (Task) Collection() string

func (*Task) GetCreatedAt

func (t *Task) GetCreatedAt() time.Time

func (Task) GetData

func (t Task) GetData() interface{}

GetData returns the data associated with the task.

func (Task) GetError

func (t Task) GetError() string

GetError returns the error that occured on the last try, or nil if none.

func (Task) GetFinishedAt

func (t Task) GetFinishedAt() *time.Time

FinishedAt returns the time the task was finished, or zero value.

func (Task) GetLog

func (t Task) GetLog() string

Returns the log messages the last task run produced.

func (Task) GetName

func (t Task) GetName() string

GetName Returns the name of the task (see @TaskSpec).

func (*Task) GetPriority

func (t *Task) GetPriority() int

func (*Task) GetProgress

func (t *Task) GetProgress() int

func (Task) GetResult

func (t Task) GetResult() interface{}

GetResult returns the result data omitted by the task.

func (*Task) GetRunAt

func (t *Task) GetRunAt() *time.Time

func (Task) GetStartedAt

func (t Task) GetStartedAt() *time.Time

StartedAt returns a time if the task was started, or zero value otherwise.

func (Task) GetTryCount

func (t Task) GetTryCount() int

TryCount returns the number of times the task has been tried.

func (*Task) IsCancelled

func (t *Task) IsCancelled() bool

func (Task) IsComplete

func (t Task) IsComplete() bool

func (*Task) IsRunning

func (t *Task) IsRunning() bool

func (Task) IsSuccess

func (t Task) IsSuccess() bool

func (*Task) SetCreatedAt

func (t *Task) SetCreatedAt(tm time.Time)

func (*Task) SetData

func (t *Task) SetData(data interface{})

func (*Task) SetError

func (t *Task) SetError(err string)

func (*Task) SetFinishedAt

func (t *Task) SetFinishedAt(tm *time.Time)

func (*Task) SetIsCancelled

func (t *Task) SetIsCancelled(x bool)

func (*Task) SetIsComplete

func (t *Task) SetIsComplete(flag bool)

func (*Task) SetIsRunning

func (t *Task) SetIsRunning(x bool)

func (*Task) SetIsSuccess

func (t *Task) SetIsSuccess(flag bool)

func (*Task) SetLog

func (t *Task) SetLog(log string)

func (*Task) SetName

func (t *Task) SetName(name string)

func (*Task) SetPriority

func (t *Task) SetPriority(x int)

func (*Task) SetProgress

func (t *Task) SetProgress(x int)

func (*Task) SetResult

func (t *Task) SetResult(result interface{})

SetResult sets the result data omitted by the task.

func (*Task) SetRunAt

func (t *Task) SetRunAt(x *time.Time)

func (*Task) SetStartedAt

func (t *Task) SetStartedAt(tm *time.Time)

func (*Task) SetTryCount

func (t *Task) SetTryCount(count int)

type TaskIntId

type TaskIntId struct {
	db.IntIdModel
	Task

	UserId uint64
}

func (TaskIntId) GetUserId

func (t TaskIntId) GetUserId() interface{}

func (TaskIntId) SetUserId

func (t TaskIntId) SetUserId(id interface{})

type TaskSpec

type TaskSpec struct {
	Name              string
	AllowedRetries    int
	RetryInterval     time.Duration
	Handler           kit.TaskHandler
	OnCompleteHandker kit.TaskOnCompleteHandler
}

func (TaskSpec) GetAllowedRetries

func (s TaskSpec) GetAllowedRetries() int

GetAllowedRetries returns the number of allowed retries.

func (TaskSpec) GetHandler

func (s TaskSpec) GetHandler() kit.TaskHandler

GetHandler returns the TaskHandler function that will execute the task.

func (TaskSpec) GetName

func (s TaskSpec) GetName() string

GetName returns a unique name for the task.

func (TaskSpec) GetOnCompleteHandler

func (s TaskSpec) GetOnCompleteHandler() kit.TaskOnCompleteHandler

func (TaskSpec) GetRetryInterval

func (s TaskSpec) GetRetryInterval() time.Duration

GetRetryInterval returns the time in seconds that must pass before a retry is attempted.

type TaskStrId

type TaskStrId struct {
	db.StrIdModel
	Task

	UserId string
}

func (TaskStrId) GetUserId

func (t TaskStrId) GetUserId() interface{}

func (TaskStrId) SetUserId

func (t TaskStrId) SetUserId(id interface{})