job

Documentation

Index

• type ID
• type Job
• type JobFunc
• type Queue
  • func NewQueue(stop <-chan struct{}, wg *sync.WaitGroup) *Queue
  • func (q *Queue) Enqueue(j *Job)
  • func (q *Queue) ForEach(fn func(int, *Job) bool)
  • func (q *Queue) Len() int
  • func (q *Queue) Ready() <-chan *Job
  • func (q *Queue) Sync()
• type Status
  • func (s Status) Error() string
• type StatusCache
  • func (c *StatusCache) SetStatus(id ID, status Status)
  • func (c *StatusCache) Status(id ID) (Status, bool)
• type StatusString

Types

type ID

type ID string

type Job

type Job struct {
	ID ID
	Do JobFunc
}

type JobFunc

type JobFunc func(log.Logger) error

type Queue

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

Queue is an unbounded queue of jobs; enqueuing a job will always proceed, while dequeuing is done by receiving from a channel. It is also possible to iterate over the current list of jobs.

func NewQueue

func NewQueue(stop <-chan struct{}, wg *sync.WaitGroup) *Queue

func (*Queue) Enqueue

func (q *Queue) Enqueue(j *Job)

Enqueue puts a job onto the queue. It will block until the queue's loop can accept the job; but this does _not_ depend on a job being dequeued and will always proceed eventually.

func (*Queue) ForEach

func (q *Queue) ForEach(fn func(int, *Job) bool)

func (*Queue) Len

func (q *Queue) Len() int

This is not guaranteed to be up-to-date; i.e., it is possible to receive from `q.Ready()` or enqueue an item, then see the same length as before, temporarily.

func (*Queue) Ready

func (q *Queue) Ready() <-chan *Job

Ready returns a channel that can be used to dequeue items. Note that dequeuing is not atomic: you may still see the dequeued item with ForEach, for a time.

func (*Queue) Sync

func (q *Queue) Sync()

Block until any previous operations have completed. Note that this is only meaningful if you are using the queue from a single other goroutine; i.e., it makes sense to do, say,

q.Enqueue(j)
q.Sync()
fmt.Printf("Queue length is %d\n", q.Len())

but only because those statements are sequential in a single thread. So this is really only useful for testing.

type Status

type Status struct {
	Result       event.CommitEventMetadata
	Err          string
	StatusString StatusString
}

Status holds the possible states of a job; either,

1. queued or otherwise pending
2. succeeded with a job-specific result
3. failed, resulting in an error and possibly a job-specific result

func (Status) Error

func (s Status) Error() string

type StatusCache

type StatusCache struct {
	// Size is the number of statuses to store. When full, jobs are evicted in FIFO ordering.
	// oldest ones will be evicted to make room.
	Size int

	sync.RWMutex
	// contains filtered or unexported fields
}

func (*StatusCache) SetStatus

func (c *StatusCache) SetStatus(id ID, status Status)

func (*StatusCache) Status

func (c *StatusCache) Status(id ID) (Status, bool)

type StatusString

type StatusString string

const (
	StatusQueued    StatusString = "queued"
	StatusRunning   StatusString = "running"
	StatusFailed    StatusString = "failed"
	StatusSucceeded StatusString = "succeeded"
)