axe

Documentation

Index

• func AddJobIndexes(indexer *coal.Indexer, removeAfter time.Duration)
• type Error
  • func E(reason string, retry bool) *Error
  • func (c *Error) Error() string
• type Job
  • func Enqueue(store *coal.SubStore, name string, data Model, delay time.Duration) (*Job, error)
• type Model
• type Pool
  • func NewPool() *Pool
  • func (p *Pool) Add(task *Task)
  • func (p *Pool) Close()
  • func (p *Pool) Run()
• type Queue
  • func NewQueue(store *coal.Store) *Queue
  • func (q *Queue) Callback(name string, delay time.Duration, matcher fire.Matcher, ...) *fire.Callback
  • func (q *Queue) Enqueue(name string, data Model, delay time.Duration) (*Job, error)
• type Status
• type Task

Functions

func AddJobIndexes

func AddJobIndexes(indexer *coal.Indexer, removeAfter time.Duration)

AddJobIndexes will add job indexes to the specified indexer. If removeAfter is specified, completed and cancelled jobs are automatically removed when their finished timestamp falls behind the specified duration.

Note: It is recommended to create custom indexes that support the exact nature of data and access patterns.

Types

type Error

type Error struct {
	Reason string
	Retry  bool
}

Error is used to signal failed job executions.

func E

func E(reason string, retry bool) *Error

E is a short-hand to construct an error.

func (*Error) Error

func (c *Error) Error() string

Error implements the error interface.

type Job

type Job struct {
	coal.Base `json:"-" bson:",inline" coal:"jobs"`

	// The name of the job.
	Name string `json:"name" bson:"name"`

	// The data that has been supplied on creation.
	Data bson.Raw `json:"data" bson:"data"`

	// The current status of the job.
	Status Status `json:"status" bson:"status"`

	// The time when the job was created.
	Created time.Time `json:"created-at" bson:"created_at"`

	// The time when the job is available for execution.
	Available time.Time `json:"available-at" bson:"available_at"`

	// The time when the job was dequeue the last time.
	Started *time.Time `json:"started-at" bson:"started_at"`

	// The time when the last attempt ended (completed, failed or cancelled).
	Ended *time.Time `json:"ended-at" bson:"ended_at"`

	// The time when the job was finished (completed or cancelled).
	Finished *time.Time `json:"finished-at" bson:"finished_at"`

	// Attempts is incremented with each execution attempt.
	Attempts int `json:"attempts" bson:"attempts"`

	// The result submitted during completion.
	Result bson.M `json:"result" bson:"result"`

	// The last message submitted when the job was failed or cancelled.
	Reason string `json:"reason" bson:"reason"`
}

Job is a single job managed by a queue.

func Enqueue

func Enqueue(store *coal.SubStore, name string, data Model, delay time.Duration) (*Job, error)

Enqueue will enqueue a job using the specified name and data. If a delay is specified the job will not be dequeued until the specified time has passed.

type Model

type Model interface{}

Model can be any BSON serializable type.

type Pool

type Pool struct {

	// The function gets invoked by the pool with critical errors.
	Reporter func(error)
	// contains filtered or unexported fields
}

Pool manages tasks and queues.

func NewPool

func NewPool() *Pool

NewPool creates and returns a new pool.

func (*Pool) Add

func (p *Pool) Add(task *Task)

Add will add the specified task and its queue to the pool.

func (*Pool) Close

func (p *Pool) Close()

Close will close the pool.

func (*Pool) Run

func (p *Pool) Run()

Run will launch the queue watchers and task workers in the background.

type Queue

type Queue struct {
	// MaxLag defines the maximum amount of lag that should be applied to every
	// dequeue attempt.
	//
	// By default multiple processes compete with each other when getting jobs
	// from the same queue. An artificial lag prevents multiple simultaneous
	// dequeue attempts and allows the worker with the smallest lag to dequeue
	// the job and inform the other processed to prevent another dequeue attempt.
	//
	// Default: 0.
	MaxLag time.Duration
	// contains filtered or unexported fields
}

Queue manages the queueing of jobs.

func NewQueue

func NewQueue(store *coal.Store) *Queue

NewQueue creates and returns a new queue.

func (*Queue) Callback

func (q *Queue) Callback(name string, delay time.Duration, matcher fire.Matcher, cb func(ctx *fire.Context) Model) *fire.Callback

Callback is a factory to create callbacks that can be used to enqueue jobs during request processing.

func (*Queue) Enqueue

func (q *Queue) Enqueue(name string, data Model, delay time.Duration) (*Job, error)

Enqueue will enqueue a job using the specified name and data. If a delay is specified the job will not dequeued until the specified time has passed.

type Status

type Status string

Status defines the allowed statuses of a job.

const (
	// StatusEnqueued is used as the initial status when jobs are created.
	StatusEnqueued Status = "enqueued"

	// StatusDequeued is set when a job has been successfully dequeued.
	StatusDequeued Status = "dequeued"

	// StatusCompleted is set when a jobs has been successfully executed.
	StatusCompleted Status = "completed"

	// StatusFailed is set when an execution of a job failed.
	StatusFailed Status = "failed"

	// StatusCancelled is set when a jobs has been cancelled.
	StatusCancelled Status = "cancelled"
)

The available job statuses.

type Task

type Task struct {
	// Name is the unique name of the task.
	Name string

	// Model is the model that holds task related data.
	Model Model

	// Queue is the queue that is used to manage the jobs.
	Queue *Queue

	// Handler is the callback called with jobs for processing. The handler
	// should return errors formatted with E to properly indicate the status of
	// the job. If a task execution is successful the handler might return some
	// data that is attached to the job.
	Handler func(Model) (bson.M, error)

	// Workers defines the number for spawned workers that dequeue and execute
	// jobs.
	//
	// Default: 2.
	Workers int

	// MaxAttempts defines the maximum attempts to complete a task.
	//
	// Default: 3
	MaxAttempts int

	// Interval defines the rate at which the worker will request a job from the
	// queue.
	//
	// Default: 100ms.
	Interval time.Duration

	// Delay is the time after a failed task is retried.
	//
	// Default: 1s.
	Delay time.Duration

	// Timeout is the time after which a task can be dequeue again in case the
	// worker was not able to set its status.
	//
	// Default: 10m.
	Timeout time.Duration
}

Task describes work that is managed using a job queue.