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 Exec(action func() error) chan error
- func Recover(errors chan error, action func() error)
- func RecoverGroup(wg *sync.WaitGroup, errors chan error, action func() error)
- func RunToError(fns ...func() error) error
- type AutoAction
- func (a *AutoAction) Increment()
- func (a *AutoAction) MaxCount() int32
- func (a *AutoAction) NotifyStarted() <-chan struct{}
- func (a *AutoAction) NotifyStopped() <-chan struct{}
- func (a *AutoAction) ShouldTriggerOnAbort() bool
- func (a *AutoAction) Start() error
- func (a *AutoAction) Stop() error
- func (a *AutoAction) Trigger()
- func (a *AutoAction) WithAction(action func()) *AutoAction
- func (a *AutoAction) WithTriggerOnAbort(triggerOnAbort bool) *AutoAction
- type AutoflushBuffer
- func (ab *AutoflushBuffer) Add(obj interface{})
- func (ab *AutoflushBuffer) AddMany(objs ...interface{})
- func (ab *AutoflushBuffer) Flush()
- func (ab *AutoflushBuffer) FlushAsync()
- func (ab *AutoflushBuffer) Interval() time.Duration
- func (ab *AutoflushBuffer) MaxLen() int
- func (ab *AutoflushBuffer) NotifyStarted() <-chan struct{}
- func (ab *AutoflushBuffer) NotifyStopped() <-chan struct{}
- func (ab *AutoflushBuffer) ShouldFlushOnAbort() bool
- func (ab *AutoflushBuffer) Start() error
- func (ab *AutoflushBuffer) Stop() error
- func (ab *AutoflushBuffer) WithFlushHandler(handler func(objs []interface{})) *AutoflushBuffer
- func (ab *AutoflushBuffer) WithFlushOnAbort(should bool) *AutoflushBuffer
- type Batch
- func (b *Batch) Abort()
- func (b *Batch) Errors() chan error
- func (b *Batch) Latch() *Latch
- func (b *Batch) NumWorkers() int
- func (b *Batch) Process()
- func (b *Batch) ProcessContext(ctx context.Context)
- func (b *Batch) WithErrors(errors chan error) *Batch
- func (b *Batch) WithNumWorkers(numWorkers int) *Batch
- func (b *Batch) WithWork(work chan interface{}) *Batch
- func (b *Batch) Work() chan interface{}
- type ErrorAction
- type ErrorWorker
- func (ew *ErrorWorker) Close() error
- func (ew *ErrorWorker) Dispatch(ctx context.Context)
- func (ew *ErrorWorker) Drain()
- func (ew *ErrorWorker) DrainContext(ctx context.Context)
- func (ew *ErrorWorker) Enqueue(obj error)
- func (ew *ErrorWorker) Execute(ctx context.Context, workItem error)
- func (ew *ErrorWorker) Latch() *Latch
- func (ew *ErrorWorker) Start()
- func (ew *ErrorWorker) StartContext(ctx context.Context)
- func (ew *ErrorWorker) Stop()
- func (ew *ErrorWorker) WithFallback(action func(error)) *ErrorWorker
- func (ew *ErrorWorker) WithWork(work chan error) *ErrorWorker
- func (ew *ErrorWorker) Work() chan error
- type Interval
- func (i *Interval) Action() func() error
- func (i *Interval) Delay() time.Duration
- func (i *Interval) Errors() chan error
- func (i Interval) Interval() time.Duration
- func (i *Interval) IsRunning() bool
- func (i *Interval) Latch() *Latch
- func (i *Interval) NotifyStarted() <-chan struct{}
- func (i *Interval) NotifyStopped() <-chan struct{}
- func (i *Interval) Start() error
- func (i *Interval) Stop() error
- func (i *Interval) WithAction(action func() error) *Interval
- func (i *Interval) WithDelay(d time.Duration) *Interval
- func (i *Interval) WithErrors(errors chan error) *Interval
- func (i *Interval) WithInterval(d time.Duration) *Interval
- type Latch
- func (l *Latch) CanStart() bool
- func (l *Latch) CanStop() bool
- func (l *Latch) IsRunning() 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()
- type ParallelQueue
- func (pq *ParallelQueue) Close() error
- func (pq *ParallelQueue) Enqueue(obj interface{})
- func (pq *ParallelQueue) Errors() chan error
- func (pq *ParallelQueue) Latch() *Latch
- func (pq *ParallelQueue) NumWorkers() int
- func (pq *ParallelQueue) Start()
- func (pq *ParallelQueue) WithErrors(errors chan error) *ParallelQueue
- func (pq *ParallelQueue) WithNumWorkers(numWorkers int) *ParallelQueue
- func (pq *ParallelQueue) WithWork(work chan interface{}) *ParallelQueue
- func (pq *ParallelQueue) Work() chan interface{}
- type QueueAction
- type Worker
- func (w *Worker) Close() error
- func (w *Worker) Dispatch(ctx context.Context)
- func (w *Worker) Drain()
- func (w *Worker) DrainContext(ctx context.Context)
- func (w *Worker) Enqueue(obj interface{})
- func (w *Worker) Errors() chan error
- func (w *Worker) Execute(ctx context.Context, workItem interface{})
- func (w *Worker) Latch() *Latch
- func (w *Worker) Start()
- func (w *Worker) StartContext(ctx context.Context)
- func (w *Worker) Stop()
- func (w *Worker) WithErrors(errors chan error) *Worker
- func (w *Worker) WithWork(work chan interface{}) *Worker
- func (w *Worker) Work() chan interface{}
Constants ¶
const ( // LatchStopped is a latch lifecycle state. LatchStopped int32 = 0 // LatchStarting is a latch lifecycle state. LatchStarting int32 = 1 // LatchRunning is a latch lifecycle state. LatchRunning int32 = 2 // LatchStopping is a latch lifecycle state. LatchStopping int32 = 3 )
const (
DefaultQueueMaxWork = 1 << 10
)
Queue constants.
Variables ¶
var ( ErrCannotStart exception.Class = "cannot start; already started" ErrCannotStop exception.Class = "cannot stop; already stopped" )
Errors
Functions ¶
func Recover ¶ added in v1.20201204.1
Recover runs an action and passes any errors to the given errors channel.
func RecoverGroup ¶ added in v1.20201204.1
RecoverGroup runs a recovery against a specific wait group with an error collector.
func RunToError ¶
RunToError runs a given set of functions until one of them panics or errors. It is useful when you need to start multiple servers and exit if any of them crashes.
Types ¶
type AutoAction ¶
AutoAction is an action that is triggered automatically on some set interval. It also exposes a function to trigger the action synchronously
func NewAutoAction ¶
func NewAutoAction(interval time.Duration, maxCount int32) *AutoAction
NewAutoAction returns a new NewAutoAction
func (*AutoAction) MaxCount ¶
func (a *AutoAction) MaxCount() int32
MaxCount returns the number of increments between action triggers
func (*AutoAction) NotifyStarted ¶
func (a *AutoAction) NotifyStarted() <-chan struct{}
NotifyStarted returns the started signal.
func (*AutoAction) NotifyStopped ¶
func (a *AutoAction) NotifyStopped() <-chan struct{}
NotifyStopped returns the started stopped.
func (*AutoAction) ShouldTriggerOnAbort ¶
func (a *AutoAction) ShouldTriggerOnAbort() bool
ShouldTriggerOnAbort refers to whether the action should be triggered when NewAutoAction is stopped
func (*AutoAction) Trigger ¶
func (a *AutoAction) Trigger()
Trigger invokes the action, if one is set, with the value This call is synchronous, in that it will call the trigger action on the same goroutine.
func (*AutoAction) WithAction ¶
func (a *AutoAction) WithAction(action func()) *AutoAction
WithAction sets the trigger action This should be called before Start(), and, if it's called after start, the AutoAction's lock should be acquired first
func (*AutoAction) WithTriggerOnAbort ¶
func (a *AutoAction) WithTriggerOnAbort(triggerOnAbort bool) *AutoAction
WithTriggerOnAbort sets whether the action should be triggered when NewAutoAction is stopped
type AutoflushBuffer ¶
type AutoflushBuffer struct {
// contains filtered or unexported fields
}
AutoflushBuffer is a backing store that operates either on a fixed length flush or a fixed interval flush. A handler should be provided but without one the buffer will just clear. Adds that would cause fixed length flushes do not block on the flush handler.
func NewAutoflushBuffer ¶
func NewAutoflushBuffer(maxLen int, interval time.Duration) *AutoflushBuffer
NewAutoflushBuffer creates a new autoflush buffer.
func (*AutoflushBuffer) Add ¶
func (ab *AutoflushBuffer) Add(obj interface{})
Add adds a new object to the buffer, blocking if it triggers a flush. If the buffer is full, it will call the flush handler on a separate goroutine.
func (*AutoflushBuffer) AddMany ¶
func (ab *AutoflushBuffer) AddMany(objs ...interface{})
AddMany adds many objects to the buffer at once.
func (*AutoflushBuffer) Flush ¶
func (ab *AutoflushBuffer) Flush()
Flush clears the buffer, if a handler is provided it is passed the contents of the buffer. This call is synchronous, in that it will call the flush handler on the same goroutine.
func (*AutoflushBuffer) FlushAsync ¶
func (ab *AutoflushBuffer) FlushAsync()
FlushAsync clears the buffer, if a handler is provided it is passed the contents of the buffer. This call is asynchronous, in that it will call the flush handler on its own goroutine.
func (*AutoflushBuffer) Interval ¶
func (ab *AutoflushBuffer) Interval() time.Duration
Interval returns the flush interval.
func (*AutoflushBuffer) MaxLen ¶
func (ab *AutoflushBuffer) MaxLen() int
MaxLen returns the maximum buffer length before a flush is triggered.
func (*AutoflushBuffer) NotifyStarted ¶ added in v0.3.1
func (ab *AutoflushBuffer) NotifyStarted() <-chan struct{}
NotifyStarted returns the started signal.
func (*AutoflushBuffer) NotifyStopped ¶ added in v0.3.1
func (ab *AutoflushBuffer) NotifyStopped() <-chan struct{}
NotifyStopped returns the started stopped.
func (*AutoflushBuffer) ShouldFlushOnAbort ¶
func (ab *AutoflushBuffer) ShouldFlushOnAbort() bool
ShouldFlushOnAbort returns if the buffer will do one final flush on abort.
func (*AutoflushBuffer) Start ¶
func (ab *AutoflushBuffer) Start() error
Start starts the buffer flusher.
func (*AutoflushBuffer) Stop ¶
func (ab *AutoflushBuffer) Stop() error
Stop stops the buffer flusher.
func (*AutoflushBuffer) WithFlushHandler ¶
func (ab *AutoflushBuffer) WithFlushHandler(handler func(objs []interface{})) *AutoflushBuffer
WithFlushHandler sets the buffer flush handler and returns a reference to the buffer.
func (*AutoflushBuffer) WithFlushOnAbort ¶
func (ab *AutoflushBuffer) WithFlushOnAbort(should bool) *AutoflushBuffer
WithFlushOnAbort sets if we should flush on aborts or not. This defaults to true.
type Batch ¶ added in v1.20201204.1
type Batch struct {
// contains filtered or unexported fields
}
Batch is a batch of work executed by a fixed count of workers.
func NewBatch ¶ added in v1.20201204.1
func NewBatch(action QueueAction, items ...interface{}) *Batch
NewBatch creates a new batch processor.
func (*Batch) NumWorkers ¶
NumWorkers returns the number of worker route
func (*Batch) Process ¶ added in v1.20201204.1
func (b *Batch) Process()
Process exeuctes the action for all the work items.
func (*Batch) ProcessContext ¶
ProcessContext exeuctes the action for all the work items.
func (*Batch) WithErrors ¶
WithErrors sets the error channel.
func (*Batch) WithNumWorkers ¶
WithNumWorkers sets the number of workers. It defaults to `runtime.NumCPU()`
type ErrorAction ¶
ErrorAction is an action handler for a queue.
type ErrorWorker ¶
type ErrorWorker struct {
// contains filtered or unexported fields
}
ErrorWorker is a worker that is pushed work as errors over a channel.
func NewErrorWorker ¶
func NewErrorWorker(action ErrorAction) *ErrorWorker
NewErrorWorker creates a new error worker.
func (*ErrorWorker) Dispatch ¶
func (ew *ErrorWorker) Dispatch(ctx context.Context)
Dispatch starts the listen loop for work.
func (*ErrorWorker) Drain ¶
func (ew *ErrorWorker) Drain()
Drain stops the worker and synchronously finishes work.
func (*ErrorWorker) DrainContext ¶
func (ew *ErrorWorker) DrainContext(ctx context.Context)
DrainContext stops the worker and synchronously drains the the remaining work with a given context.
func (*ErrorWorker) Enqueue ¶
func (ew *ErrorWorker) Enqueue(obj error)
Enqueue adds an item to the error work queue.
func (*ErrorWorker) Execute ¶
func (ew *ErrorWorker) Execute(ctx context.Context, workItem error)
Execute invokes the action and recovers panics.
func (*ErrorWorker) StartContext ¶
func (ew *ErrorWorker) StartContext(ctx context.Context)
StartContext starts the worker with a given context.
func (*ErrorWorker) Stop ¶
func (ew *ErrorWorker) Stop()
Stop stop the worker. The work left in the queue will remain.
func (*ErrorWorker) WithFallback ¶
func (ew *ErrorWorker) WithFallback(action func(error)) *ErrorWorker
WithFallback sets the fallback collector.
func (*ErrorWorker) WithWork ¶
func (ew *ErrorWorker) WithWork(work chan error) *ErrorWorker
WithWork sets the work channel. It allows you to override the default (non-buffered) channel with a buffer of your chosing.
type Interval ¶
type Interval struct {
// contains filtered or unexported fields
}
Interval is a managed goroutine that does things.
func NewInterval ¶
NewInterval returns a new worker that runs an action on an interval.
func (*Interval) NotifyStarted ¶ added in v0.3.1
func (i *Interval) NotifyStarted() <-chan struct{}
NotifyStarted returns the notify started signal.
func (*Interval) NotifyStopped ¶ added in v0.3.1
func (i *Interval) NotifyStopped() <-chan struct{}
NotifyStopped returns the notify stopped signal.
func (*Interval) WithAction ¶
WithAction sets the interval action.
func (*Interval) WithErrors ¶
WithErrors returns the error channel.
type Latch ¶
Latch is a helper to coordinate goroutine lifecycles. The lifecycle is generally as follows. 0 - stopped / idle 1 - starting 2 - running 3 - stopping goto 0 Each state includes a transition notification, i.e. `Starting()` triggers `NotifyStarting`
func (*Latch) IsStarting ¶
IsStarting indicates the latch is waiting to be scheduled.
func (*Latch) IsStopping ¶
IsStopping returns if the latch is waiting to finish stopping.
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.
func (*Latch) NotifyStarting ¶
func (l *Latch) NotifyStarting() (notifyStarting <-chan struct{})
NotifyStarting returns the started signal. It is used to coordinate the transition from stopped -> starting.
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.
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.
func (*Latch) Started ¶
func (l *Latch) Started()
Started signals that the latch is started and has entered the `IsRunning` state.
type ParallelQueue ¶
type ParallelQueue struct {
// contains filtered or unexported fields
}
ParallelQueue is a queude with multiple workers..
func NewParallelQueue ¶
func NewParallelQueue(action QueueAction) *ParallelQueue
NewParallelQueue returns a new parallel queue worker.
func (*ParallelQueue) Close ¶
func (pq *ParallelQueue) Close() error
Close stops the queue. Any work left in the queue will be discarded.
func (*ParallelQueue) Enqueue ¶
func (pq *ParallelQueue) Enqueue(obj interface{})
Enqueue adds an item to the work queue.
func (*ParallelQueue) Errors ¶
func (pq *ParallelQueue) Errors() chan error
Errors returns a channel to read action errors from. You must provide it with `WithErrors`.
func (*ParallelQueue) Latch ¶
func (pq *ParallelQueue) Latch() *Latch
Latch returns the worker latch.
func (*ParallelQueue) NumWorkers ¶
func (pq *ParallelQueue) NumWorkers() int
NumWorkers returns the number of worker route
func (*ParallelQueue) WithErrors ¶
func (pq *ParallelQueue) WithErrors(errors chan error) *ParallelQueue
WithErrors sets the error channel.
func (*ParallelQueue) WithNumWorkers ¶
func (pq *ParallelQueue) WithNumWorkers(numWorkers int) *ParallelQueue
WithNumWorkers sets the number of workers. It defaults to `runtime.NumCPU()`
func (*ParallelQueue) WithWork ¶
func (pq *ParallelQueue) WithWork(work chan interface{}) *ParallelQueue
WithWork sets the work channel.
func (*ParallelQueue) Work ¶
func (pq *ParallelQueue) Work() chan interface{}
Work returns the work channel.
type QueueAction ¶
QueueAction is an action handler for a queue.
type Worker ¶ added in v1.20201204.1
type Worker struct {
// contains filtered or unexported fields
}
Worker is a worker that is pushed work over a channel.
func NewWorker ¶ added in v1.20201204.1
func NewWorker(action QueueAction) *Worker
NewWorker creates a new worker.
func (*Worker) Drain ¶ added in v1.20201204.1
func (w *Worker) Drain()
Drain stops the worker and synchronously finishes work.
func (*Worker) DrainContext ¶
DrainContext stops the worker and synchronously drains the the remaining work with a given context.
func (*Worker) Enqueue ¶ added in v1.20201204.1
func (w *Worker) Enqueue(obj interface{})
Enqueue adds an item to the work queue.
func (*Worker) StartContext ¶
StartContext starts the worker with a given context.
func (*Worker) Stop ¶ added in v1.20201204.1
func (w *Worker) Stop()
Stop stop the worker. The work left in the queue will remain.
func (*Worker) WithErrors ¶
WithErrors returns the error channel.
Source Files