task

Documentation

Index

• Variables
• func BuildIndexName(queue queue.Name) common.IndexName
• func NewService(client *elasticsearch.Client, settings config.Tasks) task.Service
• type EsService
  • func (e *EsService) ArchiveOldTasks(ctx context.Context, archiveCompletedBefore task.CompletedAt, scrollSize uint, ...) error
  • func (e *EsService) Claim(ctx context.Context, workerId worker.Id, queues []queue.Name, ...) ([]task.Task, error)
  • func (e *EsService) Create(ctx context.Context, newTask *task.NewTask) (*task.Task, error)
  • func (e *EsService) Get(ctx context.Context, queue queue.Name, taskId task.Id) (*task.Task, error)
  • func (e *EsService) MarkDone(ctx context.Context, workerId worker.Id, queue queue.Name, taskId task.Id, ...) (*task.Task, error)
  • func (e *EsService) MarkFailed(ctx context.Context, workerId worker.Id, queue queue.Name, taskId task.Id, ...) (*task.Task, error)
  • func (e *EsService) OutstandingTasksCount(ctx context.Context, queue queue.Name, recurringTaskId task.RecurringTaskId) (uint, error)
  • func (e *EsService) ReapTimedOutTasks(ctx context.Context, scrollSize uint, scrollTtl time.Duration) error
  • func (e *EsService) RefreshAsNeeded(ctx context.Context, name queue.Name) error
  • func (e *EsService) ReportIn(ctx context.Context, workerId worker.Id, queue queue.Name, taskId task.Id, ...) (*task.Task, error)
  • func (e *EsService) SetUTCGetter(getter func() time.Time)
  • func (e *EsService) UnClaim(ctx context.Context, workerId worker.Id, queue queue.Name, taskId task.Id) (*task.Task, error)
• type PersistedTaskData
  • func ToPersistedTask(task *task.Task) PersistedTaskData

Variables

var TasquesArchiveIndex = ".tasques_archive"

var TasquesQueuePrefix = ".tasques_queue-"

Functions

func BuildIndexName

func BuildIndexName(queue queue.Name) common.IndexName

func NewService

func NewService(client *elasticsearch.Client, settings config.Tasks) task.Service

Types

type EsService

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

func (*EsService) ArchiveOldTasks

func (e *EsService) ArchiveOldTasks(ctx context.Context, archiveCompletedBefore task.CompletedAt, scrollSize uint, scrollTtl time.Duration) error

func (*EsService) Claim

func (e *EsService) Claim(ctx context.Context, workerId worker.Id, queues []queue.Name, desiredTasks uint, blockFor time.Duration) ([]task.Task, error)

func (*EsService) Create

func (e *EsService) Create(ctx context.Context, newTask *task.NewTask) (*task.Task, error)

func (*EsService) Get

func (e *EsService) Get(ctx context.Context, queue queue.Name, taskId task.Id) (*task.Task, error)

func (*EsService) MarkDone

func (e *EsService) MarkDone(ctx context.Context, workerId worker.Id, queue queue.Name, taskId task.Id, success *task.Success) (*task.Task, error)

func (*EsService) MarkFailed

func (e *EsService) MarkFailed(ctx context.Context, workerId worker.Id, queue queue.Name, taskId task.Id, failure *task.Failure) (*task.Task, error)

func (*EsService) OutstandingTasksCount

func (e *EsService) OutstandingTasksCount(ctx context.Context, queue queue.Name, recurringTaskId task.RecurringTaskId) (uint, error)

func (*EsService) ReapTimedOutTasks

func (e *EsService) ReapTimedOutTasks(ctx context.Context, scrollSize uint, scrollTtl time.Duration) error

func (*EsService) RefreshAsNeeded

func (e *EsService) RefreshAsNeeded(ctx context.Context, name queue.Name) error

func (*EsService) ReportIn

func (e *EsService) ReportIn(ctx context.Context, workerId worker.Id, queue queue.Name, taskId task.Id, newReport task.NewReport) (*task.Task, error)

func (*EsService) SetUTCGetter

func (e *EsService) SetUTCGetter(getter func() time.Time)

For testing

func (*EsService) UnClaim

func (e *EsService) UnClaim(ctx context.Context, workerId worker.Id, queue queue.Name, taskId task.Id) (*task.Task, error)

type PersistedTaskData

type PersistedTaskData struct {
	Id         string `json:"id"`
	Queue      string `json:"queue"`
	RetryTimes uint   `json:"retry_times"`
	// This doesn't map to the domain model 1:1 because storing _attempts_ instead of retires
	// allows us to not need to adjust the counts for timeouts, and instead issue a simple update-by-query
	RemainingAttempts uint                     `json:"remaining_attempts"`
	Kind              string                   `json:"kind"`
	State             task.State               `json:"state"`
	RunAt             time.Time                `json:"run_at"`
	ProcessingTimeout time.Duration            `json:"processing_timeout"`
	Priority          int                      `json:"priority"`
	Args              *jsonObjMap              `json:"args,omitempty"`
	Context           *jsonObjMap              `json:"context,omitempty"`
	LastEnqueuedAt    time.Time                `json:"last_enqueued_at"`
	LastClaimed       *persistedLastClaimed    `json:"last_claimed,omitempty"`
	Metadata          common.PersistedMetadata `json:"metadata"`
	RecurringTaskId   *string                  `json:"recurring_task_id,omitempty"`
}

func ToPersistedTask

func ToPersistedTask(task *task.Task) PersistedTaskData