async

Documentation

Overview

Package async provides syncronization primitives and background workers. This is a core package that is used by a lot of other packages.

Index

• Constants
• Variables
• func Recover(action func() error, errors chan error)
• func RecoverGroup(action func() error, errors chan error, wg *sync.WaitGroup)
• type Batch
  • func NewBatch(work chan interface{}, action WorkAction, options ...BatchOption) *Batch
  • func (b *Batch) Process(ctx context.Context)
• type BatchOption
  • func OptBatchErrors(errors chan error) BatchOption
  • func OptBatchParallelism(parallelism int) BatchOption
• type ContextAction
• type Interval
  • func NewInterval(action ContextAction, interval time.Duration, options ...IntervalOption) *Interval
  • func (i *Interval) Dispatch()
  • func (i *Interval) Start() error
  • func (i *Interval) Stop() error
• type IntervalOption
  • func OptIntervalContext(ctx context.Context) IntervalOption
  • func OptIntervalDelay(d time.Duration) IntervalOption
  • func OptIntervalErrors(errors chan error) IntervalOption
• type Latch
  • func NewLatch() *Latch
  • func (l *Latch) CanStart() bool
  • func (l *Latch) CanStop() bool
  • func (l *Latch) IsStarted() bool
  • func (l *Latch) IsStarting() bool
  • func (l *Latch) IsStopped() (isStopped bool)
  • func (l *Latch) IsStopping() bool
  • func (l *Latch) NotifyStarted() (notifyStarted <-chan struct{})
  • func (l *Latch) NotifyStarting() (notifyStarting <-chan struct{})
  • func (l *Latch) NotifyStopped() (notifyStopped <-chan struct{})
  • func (l *Latch) NotifyStopping() (notifyStopping <-chan struct{})
  • func (l *Latch) Reset()
  • func (l *Latch) Started()
  • func (l *Latch) Starting()
  • func (l *Latch) Stopped()
  • func (l *Latch) Stopping()
  • func (l *Latch) WaitStarted()
  • func (l *Latch) WaitStopped()
• type Queue
  • func NewQueue(action WorkAction, options ...QueueOption) *Queue
  • func (q *Queue) Background() context.Context
  • func (q *Queue) Close() error
  • func (q *Queue) Dispatch()
  • func (q *Queue) Enqueue(obj interface{})
  • func (q *Queue) ReturnWorker(ctx context.Context, worker *Worker) error
  • func (q *Queue) Start() error
  • func (q *Queue) Stop() error
• type QueueOption
  • func OptQueueContext(ctx context.Context) QueueOption
  • func OptQueueErrors(errors chan error) QueueOption
  • func OptQueueMaxWork(maxWork int) QueueOption
  • func OptQueueParallelism(parallelism int) QueueOption
• type WorkAction
• type Worker
  • func NewWorker(action WorkAction) *Worker
  • func (w *Worker) Background() context.Context
  • func (w *Worker) Dispatch()
  • func (w *Worker) Enqueue(obj interface{})
  • func (w *Worker) Execute(ctx context.Context, workItem interface{})
  • func (w *Worker) HandleError(err error)
  • func (w *Worker) NotifyStarted() <-chan struct{}
  • func (w *Worker) NotifyStopped() <-chan struct{}
  • func (w *Worker) Start() error
  • func (w *Worker) Stop() error
  • func (w *Worker) StopContext(ctx context.Context)
• type WorkerFinalizer

Constants

const (
	LatchStopped  int32 = 0
	LatchStarting int32 = 1
	LatchResuming int32 = 2
	LatchStarted  int32 = 3
	LatchActive   int32 = 4
	LatchPausing  int32 = 5
	LatchPaused   int32 = 6
	LatchStopping int32 = 7
)

Latch states

const (
	DefaultQueueMaxWork        = 1 << 10
	DefaultInterval            = 500 * time.Millisecond
	DefaultShutdownGracePeriod = 10 * time.Second
)

Constants

Variables

var (
	ErrCannotStart ex.Class = "cannot start; already started"
	ErrCannotStop  ex.Class = "cannot stop; already stopped"
)

Errors

Functions

func Recover

func Recover(action func() error, errors chan error)

Recover runs an action and passes any errors to the given errors channel.

This call blocks, if you need it to be backgrounded, you should call it like:

go Recover(action, errors)
<-errors

func RecoverGroup

func RecoverGroup(action func() error, errors chan error, wg *sync.WaitGroup)

RecoverGroup runs a recovery against a specific wait group with an error collector. It calls Recover internally.

Types

type Batch

type Batch struct {
	Action      WorkAction
	Parallelism int
	Work        chan interface{}
	Errors      chan error
}

Batch is a batch of work executed by a fixed count of workers.

func NewBatch

func NewBatch(work chan interface{}, action WorkAction, options ...BatchOption) *Batch

NewBatch creates a new batch processor. Batch processes are a known quantity of work that needs to be processed in parallel.

func (*Batch) Process

func (b *Batch) Process(ctx context.Context)

Process executes the action for all the work items.

type BatchOption

type BatchOption func(*Batch)

BatchOption is an option for the batch worker.

func OptBatchErrors

func OptBatchErrors(errors chan error) BatchOption

OptBatchErrors sets the batch worker error return channel.

func OptBatchParallelism

func OptBatchParallelism(parallelism int) BatchOption

OptBatchParallelism sets the batch worker parallelism, or the number of workers to create.

type ContextAction

type ContextAction func(ctx context.Context) error

ContextAction is an action that is given a context and returns an error.

type Interval

type Interval struct {
	*Latch
	Context  context.Context
	Interval time.Duration
	Action   ContextAction
	Delay    time.Duration
	Errors   chan error
}

Interval is a background worker that performs an action on an interval.

func NewInterval

func NewInterval(action ContextAction, interval time.Duration, options ...IntervalOption) *Interval

NewInterval returns a new worker that runs an action on an interval.

Example:

iw := NewInterval(func(ctx context.Context) error { return nil }, 500*time.Millisecond)
go iw.Start()
<-iw.Started()

func (*Interval) Dispatch

func (i *Interval) Dispatch()

Dispatch is the main dispatch loop.

func (*Interval) Start

func (i *Interval) Start() error

Start starts the worker.

This will start the internal ticker, with a default initial delay of the given interval, and will return an ErrCannotStart if the interval worker is already started.

This call will block.

func (*Interval) Stop

func (i *Interval) Stop() error

Stop stops the worker.

type IntervalOption

type IntervalOption func(*Interval)

IntervalOption is an option for the interval worker.

func OptIntervalContext

func OptIntervalContext(ctx context.Context) IntervalOption

OptIntervalContext sets the interval worker context.

func OptIntervalDelay

func OptIntervalDelay(d time.Duration) IntervalOption

OptIntervalDelay sets the interval worker start delay.

func OptIntervalErrors

func OptIntervalErrors(errors chan error) IntervalOption

OptIntervalErrors sets the interval worker start error channel.

type Latch

type Latch struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Latch is a helper to coordinate goroutine lifecycles, specifically waiting for goroutines to start and end.

The lifecycle is generally as follows:

0 - stopped - goto 1
1 - starting - goto 2
2 - started - goto 3
3 - stopping - goto 0

Control flow is coordinated with chan struct{}, which acts as a semaphore but can only alert (1) listener as it is buffered.

In order to start a `stopped` latch, you must call `.Reset()` first to initialize channels.

func NewLatch

func NewLatch() *Latch

NewLatch creates a new latch.

func (*Latch) CanStart

func (l *Latch) CanStart() bool

CanStart returns if the latch can start.

func (*Latch) CanStop

func (l *Latch) CanStop() bool

CanStop returns if the latch can stop.

func (*Latch) IsStarted

func (l *Latch) IsStarted() bool

IsStarted returns if the latch state is LatchStarted.

func (*Latch) IsStarting

func (l *Latch) IsStarting() bool

IsStarting returns if the latch state is LatchStarting

func (*Latch) IsStopped

func (l *Latch) IsStopped() (isStopped bool)

IsStopped returns if the latch state is LatchStopped.

func (*Latch) IsStopping

func (l *Latch) IsStopping() bool

IsStopping returns if the latch state is LatchStopping.

func (*Latch) NotifyStarted

func (l *Latch) NotifyStarted() (notifyStarted <-chan struct{})

NotifyStarted returns the started signal. It is used to coordinate the transition from starting -> started. There can only be (1) effective listener at a time for these events.

func (*Latch) NotifyStarting

func (l *Latch) NotifyStarting() (notifyStarting <-chan struct{})

NotifyStarting returns the starting signal. It is used to coordinate the transition from stopped -> starting. There can only be (1) effective listener at a time for these events.

func (*Latch) NotifyStopped

func (l *Latch) NotifyStopped() (notifyStopped <-chan struct{})

NotifyStopped returns the stopped signal. It is used to coordinate the transition from stopping -> stopped. There can only be (1) effective listener at a time for these events.

func (*Latch) NotifyStopping

func (l *Latch) NotifyStopping() (notifyStopping <-chan struct{})

NotifyStopping returns the should stop signal. It is used to trigger the transition from running -> stopping -> stopped. There can only be (1) effective listener at a time for these events.

func (*Latch) Reset

func (l *Latch) Reset()

Reset resets the latch.

func (*Latch) Started

func (l *Latch) Started()

Started signals that the latch is started and has entered the `IsStarted` state.

func (*Latch) Starting

func (l *Latch) Starting()

Starting signals the latch is starting. This is typically done before you kick off a goroutine.

func (*Latch) Stopped

func (l *Latch) Stopped()

Stopped signals the latch has stopped.

func (*Latch) Stopping

func (l *Latch) Stopping()

Stopping signals the latch to stop. It could also be thought of as `SignalStopping`.

func (*Latch) WaitStarted

func (l *Latch) WaitStarted()

WaitStarted triggers `Starting` and waits for the `Started` signal.

func (*Latch) WaitStopped

func (l *Latch) WaitStopped()

WaitStopped triggers `Stopping` and waits for the `Stopped` signal.

type Queue

type Queue struct {
	*Latch

	Action              WorkAction
	Context             context.Context
	Errors              chan error
	Parallelism         int
	MaxWork             int
	ShutdownGracePeriod time.Duration

	// these will typically be set by Start
	AvailableWorkers chan *Worker
	Workers          []*Worker
	Work             chan interface{}
}

Queue is a queue with multiple workers.

func NewQueue

func NewQueue(action WorkAction, options ...QueueOption) *Queue

NewQueue returns a new parallel queue.

func (*Queue) Background

func (q *Queue) Background() context.Context

Background returns a background context.

func (*Queue) Close

func (q *Queue) Close() error

Close stops the queue. Any work left in the queue will be discarded.

func (*Queue) Dispatch

func (q *Queue) Dispatch()

Dispatch processes work items in a loop.

func (*Queue) Enqueue

func (q *Queue) Enqueue(obj interface{})

Enqueue adds an item to the work queue.

func (*Queue) ReturnWorker

func (q *Queue) ReturnWorker(ctx context.Context, worker *Worker) error

ReturnWorker returns a given worker to the worker queue.

func (*Queue) Start

func (q *Queue) Start() error

Start starts the queue and its workers. This call blocks.

func (*Queue) Stop

func (q *Queue) Stop() error

Stop stops the queue and processes any remaining items.

type QueueOption

type QueueOption func(*Queue)

QueueOption is an option for the queue worker.

func OptQueueContext

func OptQueueContext(ctx context.Context) QueueOption

OptQueueContext sets the queue worker context.

func OptQueueErrors

func OptQueueErrors(errors chan error) QueueOption

OptQueueErrors sets the queue worker start error channel.

func OptQueueMaxWork

func OptQueueMaxWork(maxWork int) QueueOption

OptQueueMaxWork sets the queue worker max work.

func OptQueueParallelism

func OptQueueParallelism(parallelism int) QueueOption

OptQueueParallelism sets the queue worker parallelism.

type WorkAction

type WorkAction func(context.Context, interface{}) error

WorkAction is an action handler for a queue.

type Worker

type Worker struct {
	*Latch

	Context   context.Context
	Action    WorkAction
	Finalizer WorkerFinalizer

	Errors chan error
	Work   chan interface{}
}

Worker is a worker that is pushed work over a channel. It is used by other work distribution types (i.e. queue and batch) but can also be used independently.

func NewWorker

func NewWorker(action WorkAction) *Worker

NewWorker creates a new worker.

func (*Worker) Background

func (w *Worker) Background() context.Context

Background returns the queue worker background context.

func (*Worker) Dispatch

func (w *Worker) Dispatch()

Dispatch starts the listen loop for work.

func (*Worker) Enqueue

func (w *Worker) Enqueue(obj interface{})

Enqueue adds an item to the work queue.

func (*Worker) Execute

func (w *Worker) Execute(ctx context.Context, workItem interface{})

Execute invokes the action and recovers panics.

func (*Worker) HandleError

func (w *Worker) HandleError(err error)

HandleError sends a non-nil err to the error collector if one is provided.

func (*Worker) NotifyStarted

func (w *Worker) NotifyStarted() <-chan struct{}

NotifyStarted returns the underlying latch signal.

func (*Worker) NotifyStopped

func (w *Worker) NotifyStopped() <-chan struct{}

NotifyStopped returns the underlying latch signal.

func (*Worker) Start

func (w *Worker) Start() error

Start starts the worker with a given context.

func (*Worker) Stop

func (w *Worker) Stop() error

Stop stop the worker. The work left in the queue will remain.

func (*Worker) StopContext

func (w *Worker) StopContext(ctx context.Context)

StopContext stops the worker in a given cancellation context.

type WorkerFinalizer

type WorkerFinalizer func(context.Context, *Worker) error

WorkerFinalizer is an action handler for a queue.