dequeuer

Documentation

Overview

Package dequeuer retrieves jobs from the database and does some work.

Index

• type Dequeuer
  • func (d *Dequeuer) Work(name string, wg *sync.WaitGroup)
• type Pool
  • func NewPool(name string) *Pool
  • func (p *Pool) AddDequeuer(w Worker) error
  • func (p *Pool) RemoveDequeuer(ctx context.Context) error
  • func (p *Pool) Shutdown(ctx context.Context) error
• type Pools
  • func CreatePools(w Worker, maxInitialJitter time.Duration) (Pools, error)
  • func (ps Pools) NumDequeuers() int
• type Worker

Types

type Dequeuer

type Dequeuer struct {
	ID       int
	QuitChan chan bool
	W        Worker
}

func (*Dequeuer) Work

func (d *Dequeuer) Work(name string, wg *sync.WaitGroup)

type Pool

type Pool struct {
	Dequeuers []*Dequeuer
	Name      string
	// contains filtered or unexported fields
}

A Pool contains an array of dequeuers, all of which perform work for the same models.Job.

func NewPool

func NewPool(name string) *Pool

func (*Pool) AddDequeuer

func (p *Pool) AddDequeuer(w Worker) error

AddDequeuer adds a Dequeuer to the Pool. w should be the work that the Dequeuer will do with a dequeued job.

func (*Pool) RemoveDequeuer

func (p *Pool) RemoveDequeuer(ctx context.Context) error

RemoveDequeuer removes a dequeuer from the pool and sends that dequeuer a shutdown signal.

func (*Pool) Shutdown

func (p *Pool) Shutdown(ctx context.Context) error

Shutdown all workers in the pool.

type Pools

type Pools []*Pool

func CreatePools

func CreatePools(w Worker, maxInitialJitter time.Duration) (Pools, error)

CreatePools creates job pools for all jobs in the database. The provided Worker w will be shared between all dequeuers, so it must be thread safe.

func (Pools) NumDequeuers

func (ps Pools) NumDequeuers() int

NumDequeuers returns the total number of dequeuers across all pools.

type Worker

type Worker interface {
	// DoWork does whatever work should be done with the queued
	// job. Success and failure for the job are marked by hitting
	// services.HandleStatusCallback, or POST /v1/jobs/:job-name/:job-id
	// (over HTTP).
	//
	// A good pattern is for DoWork to make a HTTP request to a downstream
	// service, and then for that service to make a HTTP callback to report
	// success or failure.
	//
	// If DoWork is unable to get the work to be done, it should call
	// HandleStatusCallback with a failed callback; errors are logged, but
	// otherwise nothing else is done with them.
	DoWork(*models.QueuedJob) error

	// Sleep returns the amount of time to sleep between failed attempts to
	// acquire a queued job. The default implementation sleeps for 20, 40, 80,
	// 160, ..., up to a maximum of 10 seconds between attempts.
	Sleep(failedAttempts uint32) time.Duration
}

A Worker does some work with a QueuedJob. Worker implementations may be shared and should be threadsafe.