mysql

Documentation

Overview

Package mysql provides the implementation of a tasq repository in MySQL

Index

• type Repository
  • func NewRepository(dataSource any, prefix string) (*Repository, error)
  • func (d *Repository) CleanTasks(ctx context.Context, cleanAge time.Duration) (int64, error)
  • func (d *Repository) CountTasks(ctx context.Context, taskStatuses []tasq.TaskStatus, ...) (int64, error)
  • func (d *Repository) DeleteTask(ctx context.Context, task *tasq.Task, safeDelete bool) error
  • func (d *Repository) Migrate(ctx context.Context) error
  • func (d *Repository) PingTasks(ctx context.Context, taskIDs []uuid.UUID, visibilityTimeout time.Duration) ([]*tasq.Task, error)
  • func (d *Repository) PollTasks(ctx context.Context, types, queues []string, visibilityTimeout time.Duration, ...) ([]*tasq.Task, error)
  • func (d *Repository) PurgeTasks(ctx context.Context, taskStatuses []tasq.TaskStatus, ...) (int64, error)
  • func (d *Repository) RegisterError(ctx context.Context, task *tasq.Task, errTask error) (*tasq.Task, error)
  • func (d *Repository) RegisterFinish(ctx context.Context, task *tasq.Task, finishStatus tasq.TaskStatus) (*tasq.Task, error)
  • func (d *Repository) RegisterStart(ctx context.Context, task *tasq.Task) (*tasq.Task, error)
  • func (d *Repository) RequeueTask(ctx context.Context, task *tasq.Task) (*tasq.Task, error)
  • func (d *Repository) ScanTasks(ctx context.Context, taskStatuses []tasq.TaskStatus, ...) ([]*tasq.Task, error)
  • func (d *Repository) SubmitTask(ctx context.Context, task *tasq.Task) (*tasq.Task, error)
• type TaskID
  • func (i *TaskID) Scan(src any) error
  • func (i TaskID) Value() (driver.Value, error)

Types

type Repository

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

Repository implements the menthods necessary for tasq to work in MySQL.

func NewRepository

func NewRepository(dataSource any, prefix string) (*Repository, error)

NewRepository creates a new MySQL Repository instance.

func (*Repository) CleanTasks

func (d *Repository) CleanTasks(ctx context.Context, cleanAge time.Duration) (int64, error)

CleanTasks removes finished tasks from the queue if their creation date is past the supplied duration.

func (*Repository) CountTasks

func (d *Repository) CountTasks(ctx context.Context, taskStatuses []tasq.TaskStatus, taskTypes, queues []string) (int64, error)

CountTasks returns the number of tasks in the queue based on the supplied filters.

func (*Repository) DeleteTask

func (d *Repository) DeleteTask(ctx context.Context, task *tasq.Task, safeDelete bool) error

DeleteTask removes the supplied task from the queue.

func (*Repository) Migrate

func (d *Repository) Migrate(ctx context.Context) error

Migrate prepares the database by adding the tasks table.

func (*Repository) PingTasks

func (d *Repository) PingTasks(ctx context.Context, taskIDs []uuid.UUID, visibilityTimeout time.Duration) ([]*tasq.Task, error)

PingTasks pings a list of tasks by their ID and extends their invisibility timestamp with the supplied timeout parameter.

func (*Repository) PollTasks

func (d *Repository) PollTasks(ctx context.Context, types, queues []string, visibilityTimeout time.Duration, ordering tasq.Ordering, pollLimit int) ([]*tasq.Task, error)

PollTasks polls for available tasks matching supplied the parameters and sets their invisibility the supplied timeout parameter to the future.

func (*Repository) PurgeTasks

func (d *Repository) PurgeTasks(ctx context.Context, taskStatuses []tasq.TaskStatus, taskTypes, queues []string, safeDelete bool) (int64, error)

PurgeTasks removes all tasks from the queue based on the supplied filters.

func (*Repository) RegisterError

func (d *Repository) RegisterError(ctx context.Context, task *tasq.Task, errTask error) (*tasq.Task, error)

RegisterError records an error message on the task as last error.

func (*Repository) RegisterFinish

func (d *Repository) RegisterFinish(ctx context.Context, task *tasq.Task, finishStatus tasq.TaskStatus) (*tasq.Task, error)

RegisterFinish marks a task as finished with the supplied status and records the time of finish.

func (*Repository) RegisterStart

func (d *Repository) RegisterStart(ctx context.Context, task *tasq.Task) (*tasq.Task, error)

RegisterStart marks a task as started with the 'in progress' status and records the time of start.

func (*Repository) RequeueTask

func (d *Repository) RequeueTask(ctx context.Context, task *tasq.Task) (*tasq.Task, error)

RequeueTask marks a task as new, so it can be picked up again.

func (*Repository) ScanTasks

func (d *Repository) ScanTasks(ctx context.Context, taskStatuses []tasq.TaskStatus, taskTypes, queues []string, ordering tasq.Ordering, scanLimit int) ([]*tasq.Task, error)

ScanTasks returns a list of tasks in the queue based on the supplied filters.

func (*Repository) SubmitTask

func (d *Repository) SubmitTask(ctx context.Context, task *tasq.Task) (*tasq.Task, error)

SubmitTask adds the supplied task to the queue.

type TaskID

type TaskID [idLength]byte

TaskID represents the types used to manage conversion of UUID to MySQL's binary(16) format.

func (*TaskID) Scan

func (i *TaskID) Scan(src any) error

Scan implements sql.Scanner so TaskIDs can be read from MySQL transparently.

func (TaskID) Value

func (i TaskID) Value() (driver.Value, error)

Value implements sql.Valuer so that TaskIDs can be written to MySQL transparently.