Versions in this module Expand all Collapse all v0 v0.19.4 Nov 1, 2026 Changes in this version + var ErrClosed = errors.New("worker pool is closed") + var ErrLockNotAcquired = errors.New("lock not acquired") + var ErrQueueFull = errors.New("worker pool queue is full") + func BuildContentExtra(data map[string]interface{}) string + func Clone[T any](src *T) *T + func CopyMeta(src, dst map[string]string) + func DeepCopy(dst, src interface{}) error + func FNVHashString32(s string) uint32 + func GetBoolFromData(data map[string]interface{}, key string) bool + func GetInt64FromData(data map[string]interface{}, key string) int64 + func GetStringFromData(data map[string]interface{}, key string) string + func KvHasher[K comparable]() func(K) uint32 + func MetaMapToString(meta map[string]string) string + func MetaStringToMap(meta string) map[string]string + func MustRecover(handler RecoverFunc) + func NextPowerOfTwo(n int) int + func ParallelForEachSlice[T any](s []T, fn ParallelForEachSliceFunc[T]) + func ParallelForEach[K comparable, V any](m map[K]V, fn ParallelForEachFunc[K, V]) + func ParseContentExtraToMap(data map[string]interface{}) map[string]string + func Recover() + func RecoverAndHandle(err *error, panicHandler RecoverFunc, errorHandler func(error)) + func RecoverToError(err *error, handler RecoverFunc) + func RecoverWithHandler(handler RecoverFunc) + func SafeGo(fn func(), onPanic RecoverFunc) + func WithDefer(operation, df func()) + func WithDeferReturnValue[T any](operation func() T, df func()) T + func WithDeferReturn[T any](operation func() (T, error), df func()) (T, error) + func WithLock(lock Locker, operation func()) + func WithLockReturnFunc[R any](lock Locker, operation func() R) R + func WithLockReturnValue[T any](lock Locker, operation func() T) T + func WithLockReturnWithE[T any, E any](lock Locker, operation func() (T, E)) (T, E) + func WithLockReturn[T any](lock Locker, operation func() (T, error)) (T, error) + func WithRLock(lock RLocker, operation func()) + func WithRLockReturnFunc[R any](lock RLocker, operation func() R) R + func WithRLockReturnValue[T any](lock RLocker, operation func() T) T + func WithRLockReturnWithE[T any, E any](lock RLocker, operation func() (T, E)) (T, E) + func WithRLockReturn[T any](lock RLocker, operation func() (T, error)) (T, error) + func WithRUnlockThenLock(lock RLocker, operation func()) + func WithTryLock(lock TryLocker, operation func() error) error + func WithTryLockReturnValue[T any](lock TryLocker, operation func() T) (T, error) + func WithTryLockReturnWithE[T any, E any](lock TryLocker, operation func() (T, E)) (T, E) + func WithTryLockReturn[T any](lock TryLocker, operation func() (T, error)) (T, error) + func WithTryRLock(lock TryRLocker, operation func() error) error + func WithTryRLockReturnValue[T any](lock TryRLocker, operation func() T) (T, error) + func WithTryRLockReturnWithE[T any, E any](lock TryRLocker, operation func() (T, E)) (T, E) + func WithTryRLockReturn[T any](lock TryRLocker, operation func() (T, error)) (T, error) + func WithUnlockThenLock(lock Locker, operation func()) + type AtomicValue struct + func NewAtomicValue[T any](val T) *AtomicValue[T] + func (av *AtomicValue[T]) CompareAndSwap(old, new T) bool + func (av *AtomicValue[T]) Load() T + func (av *AtomicValue[T]) Store(val T) + func (av *AtomicValue[T]) Swap(new T) T + type BatchExecutor struct + func NewBatchExecutor(ctx context.Context) *BatchExecutor + func (b *BatchExecutor) Context() context.Context + func (b *BatchExecutor) ErrorCount() int + func (b *BatchExecutor) Errors() map[int64]error + func (b *BatchExecutor) Go(fn func() error) + func (b *BatchExecutor) HasErrors() bool + func (b *BatchExecutor) OnError(fn GoExecutorErrorHandler) *BatchExecutor + func (b *BatchExecutor) OnPanic(fn GoExecutorPanicHandler) *BatchExecutor + func (b *BatchExecutor) SetLimit(n int) *BatchExecutor + func (b *BatchExecutor) SetMode(mode BatchExecutorMode) *BatchExecutor + func (b *BatchExecutor) Wait() error + type BatchExecutorMode int + const ContinueOnErrorMode + const FailFastMode + type BatchProcessor struct + func NewBatchProcessor[T any](queueSize, batchSize int, flushInterval time.Duration, flushFn func(batch []T), ...) *BatchProcessor[T] + func (p *BatchProcessor[T]) DroppedCount() int64 + func (p *BatchProcessor[T]) Name() string + func (p *BatchProcessor[T]) Stop() + func (p *BatchProcessor[T]) Submit(item T) bool + func (p *BatchProcessor[T]) SubmitBlocking(ctx context.Context, item T) bool + type BatchProcessorOption func(*BatchProcessor[T]) + func WithBatchProcessorClone[T any](fn func(T) T) BatchProcessorOption[T] + func WithBatchProcessorName[T any](name string) BatchProcessorOption[T] + func WithBatchProcessorPanicHandler[T any](handler RecoverFunc) BatchProcessorOption[T] + type Bool struct + func NewBool(b bool) *Bool + func (b *Bool) CAS(old, new bool) bool + func (b *Bool) Load() bool + func (b *Bool) Store(val bool) + func (b *Bool) Toggle() bool + type CallbackFunc func(ctx *ExecutionContext, result T) + type ChildRunner struct + func (c *ChildRunner) Go(fn GoExecutorFunc) *ChildRunner + func (c *ChildRunner) GoWithError(fn GoExecutorWaitFunc) *ChildRunner + func (c *ChildRunner) Wait() + type Cloner interface + CloneDeep func() any + type DelayFunc func(attempt int, baseDelay time.Duration) time.Duration + type DelayStrategy int + const CustomDelayStrategy + const ExponentialDelayStrategy + const FixedDelayStrategy + const LinearDelayStrategy + const RandomDelayStrategy + type Delayer struct + func NewDelayer[T any]() *Delayer[T] + func (d *Delayer[T]) Close() + func (d *Delayer[T]) Execute() error + func (d *Delayer[T]) GetLegacyResults() []*ExecutionResult + func (d *Delayer[T]) GetResultChannel() <-chan T + func (d *Delayer[T]) GetResults() []T + func (d *Delayer[T]) GetStats() *ExecutionStats + func (d *Delayer[T]) IsRunning() bool + func (d *Delayer[T]) Stop() + func (d *Delayer[T]) Wait() error + func (d *Delayer[T]) WaitForCompletion() + func (d *Delayer[T]) WithConcurrent(concurrent bool) *Delayer[T] + func (d *Delayer[T]) WithContext(ctx context.Context) *Delayer[T] + func (d *Delayer[T]) WithCustomDelay(delayFunc DelayFunc) *Delayer[T] + func (d *Delayer[T]) WithDelay(delay time.Duration) *Delayer[T] + func (d *Delayer[T]) WithDisableCallbacks(disable bool) *Delayer[T] + func (d *Delayer[T]) WithFunction(f func() error) *Delayer[T] + func (d *Delayer[T]) WithMaxConcurrency(maxConcurrency int) *Delayer[T] + func (d *Delayer[T]) WithMaxDelay(maxDelay time.Duration) *Delayer[T] + func (d *Delayer[T]) WithMultiplier(multiplier float64) *Delayer[T] + func (d *Delayer[T]) WithOnAfterComplete(callback func(ctx *ExecutionContext)) *Delayer[T] + func (d *Delayer[T]) WithOnBeforeStart(callback func(ctx *ExecutionContext)) *Delayer[T] + func (d *Delayer[T]) WithOnComplete(callback func(result *ExecutionResult)) *Delayer[T] + func (d *Delayer[T]) WithOnError(callback func(index int, err error) bool) *Delayer[T] + func (d *Delayer[T]) WithOnErrorContext(callback ErrorHandlerFunc) *Delayer[T] + func (d *Delayer[T]) WithOnProgress(callback TaskProgressFunc) *Delayer[T] + func (d *Delayer[T]) WithOnStart(callback func(index int)) *Delayer[T] + func (d *Delayer[T]) WithOnSuccess(callback CallbackFunc[T]) *Delayer[T] + func (d *Delayer[T]) WithRandomBase(base float64) *Delayer[T] + func (d *Delayer[T]) WithSimpleFunction(f func()) *Delayer[T] + func (d *Delayer[T]) WithSimpleTaskFunc(taskFunc SimpleTaskFunc) *Delayer[T] + func (d *Delayer[T]) WithStopOnError(stopOnError bool) *Delayer[T] + func (d *Delayer[T]) WithStrategy(strategy DelayStrategy) *Delayer[T] + func (d *Delayer[T]) WithTaskFunc(taskFunc TaskFunc[T]) *Delayer[T] + func (d *Delayer[T]) WithTimes(count int) *Delayer[T] + type DynamicTicker struct + C chan time.Time + func NewDynamicTicker(interval time.Duration) *DynamicTicker + func (dt *DynamicTicker) GetInterval() time.Duration + func (dt *DynamicTicker) Start() + func (dt *DynamicTicker) Stop() + func (dt *DynamicTicker) UpdateInterval(interval time.Duration) + type ErrorHandlerFunc func(ctx *ExecutionContext) (shouldContinue bool) + type EventLoop struct + func NewEventLoop(ctx context.Context) *EventLoop + func (el *EventLoop) IfChannel(condition bool, ch interface{}, handler interface{}) *EventLoop + func (el *EventLoop) IfTicker(condition bool, interval time.Duration, handler func()) *EventLoop + func (el *EventLoop) OnChannel(ch interface{}, handler interface{}) *EventLoop + func (el *EventLoop) OnPanic(fn func(interface{})) *EventLoop + func (el *EventLoop) OnShutdown(fn func()) *EventLoop + func (el *EventLoop) OnTicker(interval time.Duration, handler func()) *EventLoop + func (el *EventLoop) Run() + func (el *EventLoop) RunAsync() + type ExecutionContext struct + Attempt int + Cancelled bool + Concurrent bool + Delay time.Duration + Duration time.Duration + EndTime time.Time + Error error + Index int + Metadata map[string]interface{} + Retryable bool + Skipped bool + StartTime time.Time + Strategy DelayStrategy + type ExecutionMode int + const Concurrent + const Sequential + type ExecutionResult struct + Duration time.Duration + EndTime time.Time + Error error + Index int + Skipped bool + StartTime time.Time + func (r *ExecutionResult) ToExecutionContext() *ExecutionContext + type ExecutionStats struct + CancelledCount int64 + EndTime time.Time + ErrorCount int64 + SkippedCount int64 + StartTime time.Time + SuccessCount int64 + TotalDuration time.Duration + type FuncChain struct + func NewFuncChain[T any]() *FuncChain[T] + func (fc *FuncChain[T]) AddFuncItem(item *FuncItem[T]) *FuncChain[T] + func (fc *FuncChain[T]) Clear() + func (fc *FuncChain[T]) Execute() + func (fc *FuncChain[T]) GetFuncItems() []FuncItem[T] + type FuncItem struct + func NewFuncItem[T any](f ReturnFunc[T]) *FuncItem[T] + func (fi *FuncItem[T]) GetError() error + func (fi *FuncItem[T]) GetResult() T + func (fi *FuncItem[T]) WithPriority(priority int) *FuncItem[T] + type GoExecutor struct + func Go(ctx ...context.Context) *GoExecutor + func GoWithContext(parent context.Context) *GoExecutor + func (g *GoExecutor) Exec(fn GoExecutorFunc) + func (g *GoExecutor) ExecSubs() + func (g *GoExecutor) ExecWithChildren(fn func(*ChildRunner)) + func (g *GoExecutor) ExecWithContext(fn GoExecutorContextFunc) + func (g *GoExecutor) ExecWithResult(fn GoExecutorResultFunc) ... + func (g *GoExecutor) OnCancel(fn GoExecutorCancelHandler) *GoExecutor + func (g *GoExecutor) OnError(fn GoExecutorErrorHandler) *GoExecutor + func (g *GoExecutor) OnPanic(fn GoExecutorPanicHandler) *GoExecutor + func (g *GoExecutor) Sub(fn GoExecutorFunc) *GoExecutor + func (g *GoExecutor) SubWithError(fn GoExecutorWaitFunc) *GoExecutor + func (g *GoExecutor) Wait(fn GoExecutorWaitFunc) error + func (g *GoExecutor) WithDelay(delay time.Duration) *GoExecutor + func (g *GoExecutor) WithTimeout(timeout time.Duration) *GoExecutor + func (g *GoExecutor) WithWaitGroup(wg *sync.WaitGroup) *GoExecutor + type GoExecutorCancelHandler func() + type GoExecutorContextFunc func(context.Context) error + type GoExecutorErrorHandler func(error) + type GoExecutorFunc func() + type GoExecutorPanicHandler = RecoverFunc + type GoExecutorResultFunc func() (interface{}, error) + type GoExecutorWaitFunc func() error + type HashedWheelTimer struct + func NewHashedWheelTimer(opts ...TimerOption) *HashedWheelTimer + func (t *HashedWheelTimer) Cancel(tt *TimerTask) + func (t *HashedWheelTimer) CancelByKey(key string) + func (t *HashedWheelTimer) Refresh(key string, delay time.Duration, task func()) *TimerTask + func (t *HashedWheelTimer) Schedule(delay time.Duration, task func()) *TimerTask + func (t *HashedWheelTimer) ScheduleWithKey(key string, delay time.Duration, task func()) *TimerTask + func (t *HashedWheelTimer) Stats() TimerStats + func (t *HashedWheelTimer) Stop() + type Int32 struct + func NewInt32(i int32) *Int32 + func (i32 *Int32) Add(i int32) int32 + func (i32 *Int32) CAS(old, new int32) bool + func (i32 *Int32) Load() int32 + func (i32 *Int32) Store(i int32) + func (i32 *Int32) String() string + func (i32 *Int32) Sub(i int32) int32 + func (i32 *Int32) Swap(i int32) int32 + type Int64 struct + func NewInt64(i int64) *Int64 + func (i64 *Int64) Add(i int64) int64 + func (i64 *Int64) CAS(old, new int64) bool + func (i64 *Int64) Load() int64 + func (i64 *Int64) Store(i int64) + func (i64 *Int64) String() string + func (i64 *Int64) Sub(i int64) int64 + func (i64 *Int64) Swap(i int64) int64 + type LimitedPool struct + func NewLimitedPool(minSize, maxSize int) *LimitedPool + func (p *LimitedPool) Get(size int) *[]byte + func (p *LimitedPool) Put(b *[]byte) + type Lock struct + func NewLock() *Lock + func (l *Lock) Lock() + func (l *Lock) TryLock() bool + func (l *Lock) Unlock() + type Locker interface + Lock func() + Unlock func() + type Map struct + func FromMap[K comparable, V comparable](data map[K]V) *Map[K, V] + func NewMap[K comparable, V comparable]() *Map[K, V] + func (m *Map[K, V]) All(predicate func(K, V) bool) bool + func (m *Map[K, V]) Any(predicate func(K, V) bool) bool + func (m *Map[K, V]) Clear() + func (m *Map[K, V]) Clone() *Map[K, V] + func (m *Map[K, V]) CompareAndDelete(key K, value V) bool + func (m *Map[K, V]) CompareAndSwap(key K, old, new V) bool + func (m *Map[K, V]) Count(predicate func(K, V) bool) int + func (m *Map[K, V]) Delete(key K) + func (m *Map[K, V]) DeleteIf(predicate func(K, V) bool) int + func (m *Map[K, V]) Equals(key K, value V, cmpFunc func(existing V) bool) bool + func (m *Map[K, V]) Filter(predicate func(K, V) bool) []V + func (m *Map[K, V]) FilterKeys(predicate func(K, V) bool) []K + func (m *Map[K, V]) FilterMap(predicate func(K, V) bool) *Map[K, V] + func (m *Map[K, V]) ForEach(fn func(K, V)) + func (m *Map[K, V]) GetOrCompute(key K, compute func() V) V + func (m *Map[K, V]) GetOrStore(key K, defaultValue V) V + func (m *Map[K, V]) IsEmpty() bool + func (m *Map[K, V]) Keys() []K + func (m *Map[K, V]) Load(key K) (V, bool) + func (m *Map[K, V]) LoadAndDelete(key K) (V, bool) + func (m *Map[K, V]) LoadOrStore(key K, new V) (V, bool) + func (m *Map[K, V]) Range(run func(key K, value V) bool) + func (m *Map[K, V]) Size() int + func (m *Map[K, V]) Store(key K, value V) + func (m *Map[K, V]) Swap(key K, value V) (pre V, ok bool) + func (m *Map[K, V]) ToMap() map[K]V + func (m *Map[K, V]) Update(key K, updater func(V) V) bool + func (m *Map[K, V]) Values() []V + type OrderedSet struct + func NewOrderedSetFromSlice[T comparable](items []T) *OrderedSet[T] + func NewOrderedSet[T comparable]() *OrderedSet[T] + func (s *OrderedSet[T]) Add(item T) + func (s *OrderedSet[T]) Clear() + func (s *OrderedSet[T]) Contains(item T) bool + func (s *OrderedSet[T]) Elements() []T + func (s *OrderedSet[T]) Len() int + func (s *OrderedSet[T]) Remove(item T) + func (s *OrderedSet[T]) String() string + type ParallelCompleteCallback func(map[K]R, map[K]error) + type ParallelEachCompleteCallback func(K) + type ParallelErrorCallback func(K, V, error) + type ParallelExecuteFunc func(K, V) (R, error) + type ParallelExecutor struct + func NewParallelExecutor[K comparable, V any, R any](m map[K]V) *ParallelExecutor[K, V, R] + func (p *ParallelExecutor[K, V, R]) Execute(fn ParallelExecuteFunc[K, V, R]) + func (p *ParallelExecutor[K, V, R]) OnComplete(fn ParallelCompleteCallback[K, R]) *ParallelExecutor[K, V, R] + func (p *ParallelExecutor[K, V, R]) OnEachComplete(fn ParallelEachCompleteCallback[K]) *ParallelExecutor[K, V, R] + func (p *ParallelExecutor[K, V, R]) OnError(fn ParallelErrorCallback[K, V]) *ParallelExecutor[K, V, R] + func (p *ParallelExecutor[K, V, R]) OnPanic(fn ParallelPanicCallback[K, V]) *ParallelExecutor[K, V, R] + func (p *ParallelExecutor[K, V, R]) OnSuccess(fn ParallelSuccessCallback[K, V, R]) *ParallelExecutor[K, V, R] + type ParallelForEachFunc func(K, V) + type ParallelForEachSliceFunc func(int, T) + type ParallelPanicCallback func(K, V, any) + type ParallelSliceCompleteCallback func([]R, []error) + type ParallelSliceEachCompleteCallback func(int) + type ParallelSliceErrorCallback func(int, T, error) + type ParallelSliceExecuteFunc func(int, T) (R, error) + type ParallelSliceExecutor struct + func NewParallelSliceExecutor[T any, R any](s []T) *ParallelSliceExecutor[T, R] + func (p *ParallelSliceExecutor[T, R]) Execute(fn ParallelSliceExecuteFunc[T, R]) + func (p *ParallelSliceExecutor[T, R]) OnComplete(fn ParallelSliceCompleteCallback[R]) *ParallelSliceExecutor[T, R] + func (p *ParallelSliceExecutor[T, R]) OnEachComplete(fn ParallelSliceEachCompleteCallback) *ParallelSliceExecutor[T, R] + func (p *ParallelSliceExecutor[T, R]) OnError(fn ParallelSliceErrorCallback[T]) *ParallelSliceExecutor[T, R] + func (p *ParallelSliceExecutor[T, R]) OnPanic(fn ParallelSlicePanicCallback[T]) *ParallelSliceExecutor[T, R] + func (p *ParallelSliceExecutor[T, R]) OnSuccess(fn ParallelSliceSuccessCallback[T, R]) *ParallelSliceExecutor[T, R] + func (p *ParallelSliceExecutor[T, R]) WithConcurrency(maxConcurrency int) *ParallelSliceExecutor[T, R] + type ParallelSlicePanicCallback func(int, T, any) + type ParallelSliceSuccessCallback func(int, T, R) + type ParallelSuccessCallback func(K, V, R) + type PeriodicTask struct + func NewPeriodicTask(name string, interval time.Duration, ...) *PeriodicTask + func (t *PeriodicTask) GetExecuteFunc() func(ctx context.Context) error + func (t *PeriodicTask) GetImmediateStart() bool + func (t *PeriodicTask) GetInterval() time.Duration + func (t *PeriodicTask) GetName() string + func (t *PeriodicTask) GetOnError() func(name string, err error) + func (t *PeriodicTask) GetOnOverlapSkipped() func(name string) + func (t *PeriodicTask) GetOnStart() func(name string) + func (t *PeriodicTask) GetOnStop() func(name string) + func (t *PeriodicTask) GetPreventOverlap() bool + func (t *PeriodicTask) IsExecuting() bool + func (t *PeriodicTask) SetImmediateStart(immediateStart bool) *PeriodicTask + func (t *PeriodicTask) SetOnError(onError func(name string, err error)) *PeriodicTask + func (t *PeriodicTask) SetOnOverlapSkipped(onOverlapSkipped func(name string)) *PeriodicTask + func (t *PeriodicTask) SetOnStart(onStart func(name string)) *PeriodicTask + func (t *PeriodicTask) SetOnStop(onStop func(name string)) *PeriodicTask + func (t *PeriodicTask) SetPreventOverlap(preventOverlap bool) *PeriodicTask + type PeriodicTaskManager struct + func NewPeriodicTaskManager() *PeriodicTaskManager + func (m *PeriodicTaskManager) AddSimpleTask(name string, interval time.Duration, ...) *PeriodicTaskManager + func (m *PeriodicTaskManager) AddTask(task *PeriodicTask) *PeriodicTaskManager + func (m *PeriodicTaskManager) AddTaskWithImmediateStart(name string, interval time.Duration, ...) *PeriodicTaskManager + func (m *PeriodicTaskManager) AddTaskWithOverlapPrevention(name string, interval time.Duration, ...) *PeriodicTaskManager + func (m *PeriodicTaskManager) AddTaskWithOverlapPreventionAndCallback(name string, interval time.Duration, ...) *PeriodicTaskManager + func (m *PeriodicTaskManager) AddTaskWithOverlapPreventionImmediateAndCallback(name string, interval time.Duration, ...) *PeriodicTaskManager + func (m *PeriodicTaskManager) ClearAllTasks() + func (m *PeriodicTaskManager) GetTaskCount() int + func (m *PeriodicTaskManager) GetTaskDetails(name ...string) []TaskDetailInfo + func (m *PeriodicTaskManager) GetTaskNames() []string + func (m *PeriodicTaskManager) IsRunning() bool + func (m *PeriodicTaskManager) RemoveTask(name string) bool + func (m *PeriodicTaskManager) RemoveTaskWithTimeout(name string, timeout time.Duration) bool + func (m *PeriodicTaskManager) SetDefaultCallbacks(onStart func(name string), onStop func(name string)) *PeriodicTaskManager + func (m *PeriodicTaskManager) SetDefaultErrorHandler(handler func(name string, err error)) *PeriodicTaskManager + func (m *PeriodicTaskManager) Start() error + func (m *PeriodicTaskManager) StartWithContext(ctx context.Context) error + func (m *PeriodicTaskManager) Stop() error + func (m *PeriodicTaskManager) StopWithTimeout(timeout time.Duration) error + func (m *PeriodicTaskManager) Wait() + func (m *PeriodicTaskManager) WaitForExecution(timeout time.Duration) error + type Pool struct + func NewPool[T any](new func() T) *Pool[T] + func (p *Pool[T]) Get() T + func (p *Pool[T]) Put(x T) + type PoolOption func(*WorkerPool) + func WithPoolIdleTimeout(d time.Duration) PoolOption + func WithWorkerPoolPanicHandler(handler RecoverFunc) PoolOption + type RLocker interface + Lock func() + RLock func() + RUnlock func() + Unlock func() + type RWLock struct + func NewRWLock() *RWLock + func (l *RWLock) Lock() + func (l *RWLock) RLock() + func (l *RWLock) RUnlock() + func (l *RWLock) TryLock() bool + func (l *RWLock) TryRLock() bool + func (l *RWLock) Unlock() + type RecoverFunc func(interface{}) + type ReturnFunc func() (T, error) + type Set struct + func NewSet[K comparable]() *Set[K] + func (s *Set[K]) Add(key K) + func (s *Set[K]) AddAll(keys ...K) + func (s *Set[K]) Clear() + func (s *Set[K]) Delete(key K) + func (s *Set[K]) DeleteAll(keys ...K) + func (s *Set[K]) Elements() []K + func (s *Set[K]) Has(key K) bool + func (s *Set[K]) HasAll(keys ...K) (existing []K, all bool) + func (s *Set[K]) IsEmpty() bool + func (s *Set[K]) Size() int + type ShardedMap struct + func NewShardedMapWithOptions[K comparable, V any](shardCount int, opts ...ShardedMapOption[K, V]) *ShardedMap[K, V] + func NewShardedMap[K comparable, V any](shardCount int) *ShardedMap[K, V] + func (m *ShardedMap[K, V]) Clear() + func (m *ShardedMap[K, V]) ClearHalf() + func (m *ShardedMap[K, V]) CompareAndDelete(key K, value V) bool + func (m *ShardedMap[K, V]) CompareAndSwap(key K, old, new V) bool + func (m *ShardedMap[K, V]) Count(filter func(K, V) bool) int + func (m *ShardedMap[K, V]) Delete(key K) + func (m *ShardedMap[K, V]) Has(key K) bool + func (m *ShardedMap[K, V]) Keys() []K + func (m *ShardedMap[K, V]) Len() int + func (m *ShardedMap[K, V]) Load(key K) (V, bool) + func (m *ShardedMap[K, V]) LoadAndDelete(key K) (V, bool) + func (m *ShardedMap[K, V]) LoadOrStore(key K, value V) (V, bool) + func (m *ShardedMap[K, V]) Range(fn func(key K, value V) bool) + func (m *ShardedMap[K, V]) RangeParallel(workerNum int, fn func(key K, value V)) + func (m *ShardedMap[K, V]) Store(key K, value V) + func (m *ShardedMap[K, V]) StoreBatch(items map[K]V) + func (m *ShardedMap[K, V]) Swap(key K, value V) (V, bool) + func (m *ShardedMap[K, V]) Values() []V + func (m *ShardedMap[K, V]) WithShardLock(key K, fn func(shardData map[K]V)) + func (m *ShardedMap[K, V]) WithShardRLock(key K, fn func(shardData map[K]V)) + type ShardedMapOption func(*shardedMapConfig) + func WithPerShardHint[K comparable, V any](perShardHint int) ShardedMapOption[K, V] + type SimpleTaskFunc func(ctx *ExecutionContext) error + type StateMachine struct + func NewStateMachine[S comparable](initialState S, opts ...StateMachineOption[S]) *StateMachine[S] + func (sm *StateMachine[S]) AllowTransition(from, to S) + func (sm *StateMachine[S]) AllowTransitions(from S, toStates ...S) + func (sm *StateMachine[S]) CanTransitionTo(to S) bool + func (sm *StateMachine[S]) ClearCallbacks() + func (sm *StateMachine[S]) ClearHistory() + func (sm *StateMachine[S]) CurrentState() S + func (sm *StateMachine[S]) DisallowTransition(from, to S) + func (sm *StateMachine[S]) GetAllowedTransitions() []S + func (sm *StateMachine[S]) GetHistory() []StateTransition[S] + func (sm *StateMachine[S]) GetHistoryCount() int + func (sm *StateMachine[S]) GetHistoryString() string + func (sm *StateMachine[S]) GetLastTransition() (StateTransition[S], bool) + func (sm *StateMachine[S]) GetLastTransitionDuration() time.Duration + func (sm *StateMachine[S]) GetLastTransitionTime() time.Time + func (sm *StateMachine[S]) GetRecentHistory(n int) []StateTransition[S] + func (sm *StateMachine[S]) GetRecentHistoryString(n int) string + func (sm *StateMachine[S]) GetTotalDuration() time.Duration + func (sm *StateMachine[S]) IsTrackingHistory() bool + func (sm *StateMachine[S]) MustTransitionTo(to S) + func (sm *StateMachine[S]) OnEnter(state S, handler func(from S)) + func (sm *StateMachine[S]) OnExit(state S, handler func(to S)) + func (sm *StateMachine[S]) OnTransition(handler func(from, to S)) + func (sm *StateMachine[S]) Reset(initialState S) + func (sm *StateMachine[S]) TransitionTo(to S) error + type StateMachineOption func(*StateMachine[S]) + func WithAllowAnyTransition[S comparable]() StateMachineOption[S] + func WithTimeFormat[S comparable](timeFormat string) StateMachineOption[S] + func WithTrackHistory[S comparable](maxHistory int) StateMachineOption[S] + type StateTransition struct + Duration time.Duration + From S + Timestamp time.Time + To S + type Task struct + func NewTaskWithOptions[T any, R any, U any](name string, fn TaskExecuteFunc[T, R], input T, ctx context.Context, ...) *Task[T, R, U] + func NewTask[T any, R any, U any](name string, fn TaskExecuteFunc[T, R], input T) *Task[T, R, U] + func (tk *Task[T, R, U]) AddDependency(dep *Task[T, R, U]) *Task[T, R, U] + func (tk *Task[T, R, U]) GetCallbackDuration() time.Duration + func (tk *Task[T, R, U]) GetCallbackError() error + func (tk *Task[T, R, U]) GetCallbackResult() U + func (tk *Task[T, R, U]) GetCallbackState() TaskState + func (tk *Task[T, R, U]) GetDependExecutionMode() ExecutionMode + func (tk *Task[T, R, U]) GetDependencyStates() map[string]TaskState + func (tk *Task[T, R, U]) GetDepends() []*Task[T, R, U] + func (tk *Task[T, R, U]) GetError() error + func (tk *Task[T, R, U]) GetFnDuration() time.Duration + func (tk *Task[T, R, U]) GetInput() T + func (tk *Task[T, R, U]) GetMaxRetries() int32 + func (tk *Task[T, R, U]) GetName() string + func (tk *Task[T, R, U]) GetResult() R + func (tk *Task[T, R, U]) GetRetryCount() int32 + func (tk *Task[T, R, U]) GetState() TaskState + func (tk *Task[T, R, U]) GetTaskHistory(taskName string) []TaskHistory + func (tk *Task[T, R, U]) GetTaskType() TaskType + func (tk *Task[T, R, U]) GetTimestamp() int64 + func (tk *Task[T, R, U]) GetTotalDuration() time.Duration + func (tk *Task[T, R, U]) SetDependExecutionMode(executionMode ExecutionMode) *Task[T, R, U] + func (tk *Task[T, R, U]) SetFailureCallback(callback TaskCallbackFunc[R, U]) *Task[T, R, U] + func (tk *Task[T, R, U]) SetMaxHistorySize(maxHistorySize int) *Task[T, R, U] + func (tk *Task[T, R, U]) SetMaxRetries(count int32) *Task[T, R, U] + func (tk *Task[T, R, U]) SetPriority(priority int) *Task[T, R, U] + func (tk *Task[T, R, U]) SetRetryInterval(retryInterval time.Duration) *Task[T, R, U] + func (tk *Task[T, R, U]) SetSuccessCallback(callback TaskCallbackFunc[R, U]) *Task[T, R, U] + func (tk *Task[T, R, U]) SetTimeout(timeout time.Duration) *Task[T, R, U] + type TaskCallbackFunc func(result R, err error) (U, error) + type TaskDetailInfo struct + ImmediateStart bool + Interval time.Duration + IsExecuting bool + Name string + PreventOverlap bool + type TaskExecuteFunc func(ctx context.Context, input T) (R, error) + type TaskFunc func(ctx *ExecutionContext) (T, error) + type TaskHistory struct + func (th *TaskHistory) GetCallbackDuration() time.Duration + func (th *TaskHistory) GetError() error + func (th *TaskHistory) GetFnDuration() time.Duration + func (th *TaskHistory) GetResult() interface{} + func (th *TaskHistory) GetState() TaskState + func (th *TaskHistory) GetTaskType() TaskType + func (th *TaskHistory) GetTimestamp() int64 + type TaskManager struct + func NewTaskManager[T any, R any, U any](concurrency int) *TaskManager[T, R, U] + func (tm *TaskManager[T, R, U]) AddTask(task *Task[T, R, U]) *TaskManager[T, R, U] + func (tm *TaskManager[T, R, U]) Cancel(taskName string) + func (tm *TaskManager[T, R, U]) CancelAll() + func (tm *TaskManager[T, R, U]) GetTask(name string) (*Task[T, R, U], bool) + func (tm *TaskManager[T, R, U]) GetTasks() map[string]*Task[T, R, U] + func (tm *TaskManager[T, R, U]) Run() error + func (tm *TaskManager[T, R, U]) SetTrunDown(fn TaskTrunFunc[R]) *TaskManager[T, R, U] + func (tm *TaskManager[T, R, U]) SetTrunUp(fn TaskTrunFunc[R]) *TaskManager[T, R, U] + func (tm *TaskManager[T, R, U]) TrunDown() (result R, err error) + func (tm *TaskManager[T, R, U]) TrunUp() (result R, err error) + type TaskProgressFunc func(completed int64, total int64, percentage float64) + type TaskState int32 + const Cancelled + const Completed + const Failed + const Pending + const Running + type TaskTrunFunc func() (R, error) + type TaskType int + const DependencyTask + const MainTask + type Timer interface + Cancel func(t *TimerTask) + CancelByKey func(key string) + Refresh func(key string, delay time.Duration, task func()) *TimerTask + Schedule func(delay time.Duration, task func()) *TimerTask + ScheduleWithKey func(key string, delay time.Duration, task func()) *TimerTask + Stats func() TimerStats + Stop func() + type TimerOption func(*timerConfig) + func WithTimerBucketCount(count int) TimerOption + func WithTimerExecutor(fn func(func())) TimerOption + func WithTimerShardCount(count int) TimerOption + func WithTimerTickInterval(d time.Duration) TimerOption + type TimerStats struct + ActiveTasks int64 + CancelledTasks int64 + CompletedTasks int64 + type TimerTask struct + type TryLocker interface + TryLock func() bool + type TryRLocker interface + TryRLock func() bool + type Uint32 struct + func NewUint32(i uint32) *Uint32 + func (u32 *Uint32) Add(i uint32) uint32 + func (u32 *Uint32) CAS(old, new uint32) bool + func (u32 *Uint32) Load() uint32 + func (u32 *Uint32) Store(i uint32) + func (u32 *Uint32) String() string + func (u32 *Uint32) Sub(i uint32) uint32 + func (u32 *Uint32) Swap(i uint32) uint32 + type Uint64 struct + func NewUint64(i uint64) *Uint64 + func (u64 *Uint64) Add(i uint64) uint64 + func (u64 *Uint64) CAS(old, new uint64) bool + func (u64 *Uint64) Load() uint64 + func (u64 *Uint64) Store(i uint64) + func (u64 *Uint64) String() string + func (u64 *Uint64) Sub(i uint64) uint64 + func (u64 *Uint64) Swap(i uint64) uint64 + type Uintptr struct + func NewUintptr(i uintptr) *Uintptr + func (ptr *Uintptr) CAS(old, new uintptr) bool + func (ptr *Uintptr) Load() uintptr + func (ptr *Uintptr) Store(i uintptr) + func (ptr *Uintptr) String() string + func (ptr *Uintptr) Swap(i uintptr) uintptr + type WorkerPool struct + func NewWorkerPool(workers, queueSize int, opts ...PoolOption) *WorkerPool + func (p *WorkerPool) Close() error + func (p *WorkerPool) GetActiveWorkerCount() int + func (p *WorkerPool) GetQueueSize() int + func (p *WorkerPool) GetWorkerCount() int + func (p *WorkerPool) IsClosed() bool + func (p *WorkerPool) Submit(ctx context.Context, task WorkerTask) error + func (p *WorkerPool) SubmitNonBlocking(task WorkerTask) error + func (p *WorkerPool) Wait() + type WorkerTask func()
Go to main page