chanque

Oct 11, 2022

Changes in this version

type Queue

+ func (q *Queue) Closed() bool

type WorkerOptionFunc

+ func WorkerAutoShutdown(enable bool) WorkerOptionFunc

+ func WorkerTimeout(timeout time.Duration) WorkerOptionFunc

Aug 31, 2022

Changes in this version

type Executor

+ func (e *Executor) CurrentStacktrace() []byte

type ExecutorOptionFunc

+ func ExecutorCollectStacktrace(enable bool) ExecutorOptionFunc

Apr 19, 2022

Changes in this version

type WorkerOptionFunc

+ func WorkerMaxDequeueSize(size int) WorkerOptionFunc

Sep 30, 2021

Changes in this version

+ type Wait struct

+ func WaitN(n int) *Wait

+ func WaitOne() *Wait

+ func WaitTimeout(dur time.Duration, n int) *Wait

+ func WaitTwo() *Wait

+ func (w *Wait) Cancel()

+ func (w *Wait) Done()

+ func (w *Wait) Wait() error

+ type WaitRendezvous struct

+ func WaitRendez(n int) *WaitRendezvous

+ func WaitRendezTimeout(dur time.Duration, n int) *WaitRendezvous

+ func (r *WaitRendezvous) Cancel()

+ func (r *WaitRendezvous) Wait() error

+ type WaitReplyFunc func(interface{}) (interface{}, error)

+ type WaitRequest struct

+ func WaitReq() *WaitRequest

+ func WaitReqTimeout(dur time.Duration) *WaitRequest

+ func (r *WaitRequest) Cancel()

+ func (r *WaitRequest) Req(v interface{}) error

+ func (r *WaitRequest) Wait() (interface{}, error)

+ type WaitRequestReply struct

+ func WaitReqReply() *WaitRequestReply

+ func WaitReqReplyTimeout(dur time.Duration) *WaitRequestReply

+ func (r *WaitRequestReply) Cancel()

+ func (r *WaitRequestReply) Reply(fn WaitReplyFunc) error

+ func (r *WaitRequestReply) Req(v interface{}) (interface{}, error)

+ type WaitSequence struct

+ func WaitSeq(wn ...*Wait) *WaitSequence

+ func WaitSeqTimeout(dur time.Duration, wn ...*Wait) *WaitSequence

+ func (w *WaitSequence) Cancel()

+ func (w *WaitSequence) Wait() error

Feb 5, 2021

Changes in this version

type WorkerOptionFunc

+ func WorkerCapacity(capacity int) WorkerOptionFunc

Jan 13, 2021

Changes in this version

+ var ErrPipeClosed = errors.New("pipe queue closed")

type PipelineOptionFunc

+ func PipelineContext(ctx context.Context) PipelineOptionFunc

+ type WorkerAbortQueueHandlerFunc func(paramter interface{})

type WorkerOptionFunc

+ func WorkerAbortQueueHandler(handler WorkerAbortQueueHandlerFunc) WorkerOptionFunc

Nov 18, 2020

Changes in this version

+ var ErrRetryContextTimeout = errors.New("Retry context timeout or canceled")

+ var ErrRetryNotComplete = errors.New("RetryFuture retry not complete")

+ var RetryRand *rand.Rand = rand.New(rand.NewSource(time.Now().UnixNano()))

+ type Backoff struct

+ func NewBackoff(min, max time.Duration) *Backoff

+ func NewBackoffNoJitter(min, max time.Duration) *Backoff

+ func (b *Backoff) Next() time.Duration

+ func (b *Backoff) Reset()

+ type Retry struct

+ func NewRetry(executor *Executor, funcs ...RetryOptionFunc) *Retry

+ func NewRetryWithBackoff(executor *Executor, backoff *Backoff, funcs ...RetryOptionFunc) *Retry

+ func (r *Retry) Retry(fn RetryFunc) *RetryFuture

+ func (r *Retry) RetryWithErrorHandler(fn RetryFunc, eh RetryErrorHandler) *RetryFuture

+ type RetryErrorHandler func(err error, b *Backoff) RetryNext

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

+ type RetryFuture struct

+ func (f *RetryFuture) Result() ValueError

+ type RetryNext uint8

+ const RetryNextBreak

+ const RetryNextContinue

+ type RetryOptionFunc func(*optRetry)

+ func RetryBackoffIntervalMax(dur time.Duration) RetryOptionFunc

+ func RetryBackoffIntervalMin(dur time.Duration) RetryOptionFunc

+ func RetryBackoffUseJitter(useJitter bool) RetryOptionFunc

+ func RetryContext(ctx context.Context) RetryOptionFunc

+ func RetryMax(max int) RetryOptionFunc

Sep 8, 2020

Changes in this version

+ type Parallel struct

+ func NewParallel(e *Executor, funcs ...ParallelOptionFunc) *Parallel

+ func (p *Parallel) Queue(fn ParallelJob)

+ func (p *Parallel) Submit() *ParallelFuture

+ type ParallelFuture struct

+ func (f *ParallelFuture) Result() []ValueError

+ type ParallelFutureResult struct

+ func (r *ParallelFutureResult) Results() []ValueError

+ type ParallelJob func() (result interface{}, err error)

+ type ParallelOptionFunc func(*optParallel)

+ func ParallelContext(ctx context.Context) ParallelOptionFunc

+ func Parallelism(p int) ParallelOptionFunc

+ type ValueError interface

+ Err func() error

+ Value func() interface{}

May 31, 2020

Changes in this version

type Executor

+ func (e *Executor) TuneMaxWorker(nextMaxWorker int)

+ func (e *Executor) TuneMinWorker(nextMinWorker int)

May 30, 2020

Changes in this version

+ type Loop struct

+ func NewLoop(e *Executor, funcs ...LoopOptionFunc) *Loop

+ func (lo *Loop) Execute()

+ func (lo *Loop) ExecuteTimeout(timeout time.Duration)

+ func (lo *Loop) SetDefault(h LoopDefaultHandler)

+ func (lo *Loop) SetDequeue(h LoopDequeueHandler, q *Queue)

+ func (lo *Loop) SetTicker(h LoopTickerHandler, dur time.Duration)

+ func (lo *Loop) Stop()

+ func (lo *Loop) StopAndWait()

+ type LoopDefaultHandler func() LoopNext

+ type LoopDequeueHandler func(val interface{}, ok bool) LoopNext

+ type LoopNext uint8

+ const LoopNextBreak

+ const LoopNextContinue

+ type LoopOptionFunc func(*optLoop)

+ func LoopContext(ctx context.Context) LoopOptionFunc

+ type LoopTickerHandler func() LoopNext

May 30, 2020

Changes in this version

type Executor

+ func (e *Executor) MaxWorker() int

+ func (e *Executor) MinWorker() int

type PipelineOptionFunc

+ func PipelineExecutor(executor *Executor) PipelineOptionFunc

type Queue

+ func (q *Queue) Cap() int

+ func (q *Queue) Len() int

May 27, 2020

Changes in this version

type WorkerOptionFunc

+ func WorkerExecutor(executor *Executor) WorkerOptionFunc

May 27, 2020

Changes in this version

+ type ContextTimeout struct

+ func NewContextTimeout(executor *Executor, done DoneFunc, timeout time.Duration) *ContextTimeout

+ func (c *ContextTimeout) Add() func()

+ func (c *ContextTimeout) Background()

+ func (c *ContextTimeout) Wait()

May 26, 2020

Changes in this version

+ type Context struct

+ func NewContext(executor *Executor, done DoneFunc) *Context

+ func (c *Context) Add() func()

+ func (c *Context) Background()

+ func (c *Context) Wait()

+ type DoneFunc func()

May 26, 2020

Changes in this version

type Executor

+ func (e *Executor) SubExecutor() *SubExecutor

+ type SubExecutor struct

+ func (s *SubExecutor) Submit(fn Job)

+ func (s *SubExecutor) Wait()

May 25, 2020

Changes in this version

+ const Version

+ func NewBufferWorker(handler WorkerHandler, funcs ...WorkerOptionFunc) *bufferWorker

+ func NewDefaultWorker(handler WorkerHandler, funcs ...WorkerOptionFunc) *defaultWorker

+ type Executor struct

+ func NewExecutor(minWorker, maxWorker int, funcs ...ExecutorOptionFunc) *Executor

+ func (e *Executor) ForceStop()

+ func (e *Executor) Release()

+ func (e *Executor) ReleaseAndWait()

+ func (e *Executor) Running() int32

+ func (e *Executor) Submit(fn Job)

+ func (e *Executor) Workers() int32

+ type ExecutorOption struct

+ type ExecutorOptionFunc func(*ExecutorOption)

+ func ExecutorContext(ctx context.Context) ExecutorOptionFunc

+ func ExecutorMaxCapacity(capacity int) ExecutorOptionFunc

+ func ExecutorPanicHandler(handler PanicHandler) ExecutorOptionFunc

+ func ExecutorReducderInterval(interval time.Duration) ExecutorOptionFunc

+ type Job func()

+ type PanicHandler func(PanicType, interface{})

+ type PanicType uint8

+ const PanicTypeClose

+ const PanicTypeDequeue

+ const PanicTypeEnqueue

+ func (p PanicType) String() string

+ type Pipeline struct

+ func NewPipeline(inFunc PipelineInputFunc, outFunc PipelineOutputFunc, ...) *Pipeline

+ func (p *Pipeline) CloseEnqueue() bool

+ func (p *Pipeline) Enqueue(parameter interface{}) bool

+ func (p *Pipeline) Shutdown()

+ func (p *Pipeline) ShutdownAndWait()

+ type PipelineInputFunc func(parameter interface{}) (result interface{}, err error)

+ type PipelineOption struct

+ type PipelineOptionFunc func(*PipelineOption)

+ func PipelinePanicHandler(handler PanicHandler) PipelineOptionFunc

+ type PipelineOutputFunc func(result interface{}, err error)

+ type Queue struct

+ func NewQueue(c int, funcs ...QueueOptionFunc) *Queue

+ func (q *Queue) Chan() <-chan interface{}

+ func (q *Queue) Close() (closed bool)

+ func (q *Queue) Dequeue() (val interface{}, found bool)

+ func (q *Queue) DequeueNB() (val interface{}, found bool)

+ func (q *Queue) DequeueRetry(retryInterval time.Duration, retryLimit int) (val interface{}, found bool)

+ func (q *Queue) Enqueue(val interface{}) (write bool)

+ func (q *Queue) EnqueueNB(val interface{}) (write bool)

+ func (q *Queue) EnqueueRetry(val interface{}, retryInterval time.Duration, retryLimit int) (write bool)

+ type QueueOption struct

+ type QueueOptionFunc func(*QueueOption)

+ func QueuePanicHandler(handler PanicHandler) QueueOptionFunc

+ type Worker interface

+ CloseEnqueue func() bool

+ Enqueue func(interface{}) bool

+ ForceStop func()

+ Shutdown func()

+ ShutdownAndWait func()

+ type WorkerHandler func(parameter interface{})

+ type WorkerHook func()

+ type WorkerOption struct

+ type WorkerOptionFunc func(*WorkerOption)

+ func WorkerContext(ctx context.Context) WorkerOptionFunc

+ func WorkerPanicHandler(handler PanicHandler) WorkerOptionFunc

+ func WorkerPostHook(hook WorkerHook) WorkerOptionFunc

+ func WorkerPreHook(hook WorkerHook) WorkerOptionFunc