varmq

package module

Go to main page

Versions in this module

v1

v1.9.0

Oct 18, 2026

Changes in this version

+ const InstanceHeader

+ var InstanceID = os.Hostname()

type IWorkerBinder

+ WithTopic func(t Topic[T], configs ...TopicConfigFunc) Subscription[T]

+ type Subscription interface

+ Unsubscribe func()

+ Worker func() Worker

+ type Topic interface

+ Close func() error

+ IsFull func() bool

+ NumSubscribers func() int

+ Publish func(data T) bool

+ func NewTopic[T any](configs ...TopicConfigFunc) Topic[T]

+ type TopicConfigFunc func(*topicConfig)

+ func WithTopicCapacity(capacity int) TopicConfigFunc

+ func WithTopicPriority(priority int) TopicConfigFunc

v1.8.0

Jul 4, 2026

Changes in this version

+ func DefaultErrHandler(cb func(Worker, error))

type ConfigFunc

+ func WithErrHandler(cb func(Worker, error)) ConfigFunc

v1.7.1

May 25, 2026

v1.7.0

May 16, 2026

Changes in this version

+ var ErrWorkerPaused = errors.New("worker is paused")

+ var ErrWorkerPausing = errors.New("worker is pausing")

+ var ErrWorkerStopped = errors.New("worker is stopped")

+ var ErrWorkerStopping = errors.New("worker is stopping")

+ var WorkerRegistry sync.Map

+ func DefaultStoppedRetention(duration time.Duration)

+ func Handler(prefix string) http.Handler

+ func NewServerMux() *http.ServeMux

type ConfigFunc

+ func WithAutoRun(run bool) ConfigFunc

+ func WithName(name string) ConfigFunc

+ func WithStoppedRetention(duration time.Duration) ConfigFunc

type Worker

+ Name func() string

+ Start func() error

+ StopAndWait func() error

+ Wait func()

+ WaitAndPause func() error

+ WaitUntilPaused func()

+ WaitUntilStopped func()

v1.6.1

Mar 13, 2026

v1.6.0

Mar 8, 2026

Changes in this version

+ type CapacitySetter interface

+ SetCapacity func(cap int)

type IExternalBaseQueue

+ IsFull func() bool

+ Len func() int

+ NumPending func() int

type QueueConfigFunc

+ func WithQueueCapacity(capacity int) QueueConfigFunc

type Worker

+ IsActive func() bool

+ IsIdle func() bool

+ WaitUntilIdle func()

v1.5.0

Feb 24, 2026

Changes in this version

+ func DefaultConcurrency(concurrency int)

+ func DefaultCtx(ctx context.Context)

+ func DefaultIdleWorkerExpiryDuration(duration time.Duration)

+ func DefaultJobIdGenerator(fn func() string)

+ func DefaultMinIdleWorkerRatio(percentage uint8)

+ func DefaultStrategy(s Strategy)

+ type QueueConfig struct

+ Priority int

+ type QueueConfigFunc func(*QueueConfig)

+ func WithQueuePriority(priority int) QueueConfigFunc

type Strategy

+ const Priority

v1.4.0

Feb 5, 2026

Changes in this version

+ var ErrAcknowledgeJob = errors.New("queue failed to acknowledge job")

+ var ErrFailedToCastJob = errors.New("failed to cast job")

+ var ErrFailedToDequeue = errors.New("failed to dequeue job")

+ var ErrGetNextQueue = errors.New("failed to get next queue")

+ var ErrJobAlreadyClosed = errors.New("job is already closed")

+ var ErrJobProcessing = errors.New("job is processing, you can't close processing job")

+ var ErrNotRunningWorker = errors.New("worker is not running")

+ var ErrParseJob = errors.New("failed to parse job")

+ var ErrRunningWorker = errors.New("worker is already running")

+ var ErrSameConcurrency = errors.New("worker already has the same concurrency")

type ConfigFunc

+ func WithContext(ctx context.Context) ConfigFunc

+ type Metrics interface

+ Completed func() uint64

+ Failed func() uint64

+ Reset func()

+ Submitted func() uint64

+ Successful func() uint64

type Worker

+ Context func() context.Context

+ Errs func() <-chan error

+ Metrics func() Metrics

+ NumPending func() int

v1.3.1

Jun 30, 2025

v1.3.0

Jun 21, 2025

Changes in this version

+ func ErrFunc() func(j Job[func() error]) error

+ func Func() func(j Job[func()])

+ func ResultFunc[R any]() func(j Job[func() (R, error)]) (R, error)

type ConfigFunc

+ func WithStrategy(s Strategy) ConfigFunc

type PersistentPriorityQueue

+ Worker func() Worker

+ type Strategy uint8

+ const MaxLen

+ const MinLen

+ const RoundRobin

v1.2.0

May 29, 2025

Changes in this version

+ type Awaitable interface

+ Wait func()

type ConfigFunc

+ func WithIdleWorkerExpiryDuration(duration time.Duration) ConfigFunc

+ func WithMinIdleWorkerRatio(percentage uint8) ConfigFunc

+ type Drainer interface

+ Drain func()

+ type EnqueuedErrGroupJob interface

+ Errs func() <-chan error

+ type EnqueuedErrJob interface

+ Err func() error

type EnqueuedJob

+ Close func() error

+ type EnqueuedResultGroupJob interface

+ Results func() <-chan Result[R]

+ type EnqueuedResultJob interface

+ Result func() (R, error)

+ type ErrPriorityQueue interface

+ Add func(data T, priority int, configs ...JobConfigFunc) (EnqueuedErrJob, bool)

+ AddAll func(data []Item[T]) EnqueuedErrGroupJob

+ Worker func() Worker

+ type ErrQueue interface

+ Add func(data T, configs ...JobConfigFunc) (EnqueuedErrJob, bool)

+ AddAll func(data []Item[T]) EnqueuedErrGroupJob

+ Worker func() Worker

+ type IErrWorkerBinder interface

+ BindPriorityQueue func() ErrPriorityQueue[T]

+ BindQueue func() ErrQueue[T]

+ WithPriorityQueue func(pq IPriorityQueue) ErrPriorityQueue[T]

+ WithQueue func(q IQueue) ErrQueue[T]

+ type IResultWorkerBinder interface

+ BindPriorityQueue func() ResultPriorityQueue[T, R]

+ BindQueue func() ResultQueue[T, R]

+ WithPriorityQueue func(pq IPriorityQueue) ResultPriorityQueue[T, R]

+ WithQueue func(q IQueue) ResultQueue[T, R]

+ func NewResultWorker[T, R any](wf func(j Job[T]) (R, error), config ...any) IResultWorkerBinder[T, R]

type IWorkerBinder

+ WithDistributedPriorityQueue func(dq IDistributedPriorityQueue) DistributedPriorityQueue[T]

+ WithDistributedQueue func(dq IDistributedQueue) DistributedQueue[T]

+ type Identifiable interface

+ ID func() string

type Item

+ Data T

+ Priority int

type Job

+ Data func() T

+ type PendingTracker interface

+ NumPending func() int

type PersistentPriorityQueue

+ Add func(data T, priority int, configs ...JobConfigFunc) bool

type PersistentQueue

+ Add func(data T, configs ...JobConfigFunc) bool

+ Worker func() Worker

type PriorityQueue

+ Worker func() Worker

type Queue

+ Worker func() Worker

+ type ResultPriorityQueue interface

+ Add func(data T, priority int, configs ...JobConfigFunc) (EnqueuedResultJob[R], bool)

+ AddAll func(data []Item[T]) EnqueuedResultGroupJob[R]

+ Worker func() Worker

+ type ResultQueue interface

+ Add func(data T, configs ...JobConfigFunc) (EnqueuedResultJob[R], bool)

+ AddAll func(data []Item[T]) EnqueuedResultGroupJob[R]

+ Worker func() Worker

+ type StatusProvider interface

+ IsClosed func() bool

+ Status func() string

type Worker

+ NumConcurrency func() int

+ NumIdleWorkers func() int

+ NumProcessing func() int

+ TunePool func(concurrency int) error

+ WaitAndStop func() error

+ WaitUntilFinished func()

v1.1.0

May 1, 2025

Changes in this version

+ type ConfigFunc func(*configs)

+ func WithAutoCleanupCache(duration time.Duration) ConfigFunc

+ func WithCache(cache ICache) ConfigFunc

+ func WithConcurrency(concurrency int) ConfigFunc

+ func WithJobIdGenerator(fn func() string) ConfigFunc

+ type DistributedPriorityQueue interface

+ Add func(data T, priority int, configs ...JobConfigFunc) bool

+ func NewDistributedPriorityQueue[T, R any](internalQueue IDistributedPriorityQueue) DistributedPriorityQueue[T, R]

+ type DistributedQueue interface

+ Add func(data T, configs ...JobConfigFunc) bool

+ func NewDistributedQueue[T, R any](internalQueue IDistributedQueue) DistributedQueue[T, R]

+ type EnqueuedGroupJob interface

+ Drain func() error

+ Results func() (<-chan Result[T], error)

+ type EnqueuedJob interface

+ Drain func() error

+ Result func() (R, error)

+ type EnqueuedSingleGroupJob interface

+ type IAcknowledgeable interface

+ Acknowledge func(ackID string) bool

+ DequeueWithAckId func() (any, bool, string)

+ type IBaseQueue interface

+ Close func() error

+ Dequeue func() (any, bool)

+ Len func() int

+ Purge func()

+ Values func() []any

+ type ICache interface

+ Clear func()

+ Delete func(key any)

+ Load func(key any) (any, bool)

+ Range func(f func(key, value any) bool)

+ Store func(key any, value any)

+ type IDistributedPriorityQueue interface

+ type IDistributedQueue interface

+ type IExternalBaseQueue interface

+ Close func() error

+ PendingCount func() int

+ Purge func()

+ type IExternalQueue interface

+ GroupsJobById func(id string) (EnqueuedSingleGroupJob[R], error)

+ JobById func(id string) (EnqueuedJob[R], error)

+ WaitAndClose func() error

+ WaitUntilFinished func()

+ Worker func() Worker[T, R]

+ type IPersistentPriorityQueue interface

+ type IPersistentQueue interface

+ type IPriorityQueue interface

+ Enqueue func(item any, priority int) bool

+ type IQueue interface

+ Enqueue func(item any) bool

+ type ISubscribable interface

+ Subscribe func(func(action string))

+ type IVoidWorkerBinder interface

+ WithDistributedPriorityQueue func(dq IDistributedPriorityQueue) DistributedPriorityQueue[T, any]

+ WithDistributedQueue func(dq IDistributedQueue) DistributedQueue[T, any]

+ func NewVoidWorker[T any](wf VoidWorkerFunc[T], config ...any) IVoidWorkerBinder[T]

+ type IWorkerBinder interface

+ BindPriorityQueue func() PriorityQueue[T, R]

+ BindQueue func() Queue[T, R]

+ WithPersistentPriorityQueue func(pq IPersistentPriorityQueue) PersistentPriorityQueue[T, R]

+ WithPersistentQueue func(pq IPersistentQueue) PersistentQueue[T, R]

+ WithPriorityQueue func(pq IPriorityQueue) PriorityQueue[T, R]

+ WithQueue func(q IQueue) Queue[T, R]

+ func NewErrWorker[T any](wf WorkerErrFunc[T], config ...any) IWorkerBinder[T, any]

+ func NewWorker[T, R any](wf WorkerFunc[T, R], config ...any) IWorkerBinder[T, R]

+ type Item struct

+ ID string

+ Value T

+ type Job interface

+ ID func() string

+ IsClosed func() bool

+ Json func() ([]byte, error)

+ Status func() string

+ type JobConfigFunc func(*jobConfigs)

+ func WithJobId(id string) JobConfigFunc

+ type PQItem struct

+ ID string

+ Priority int

+ Value T

+ type PersistentPriorityQueue interface

+ type PersistentQueue interface

+ type PriorityQueue interface

+ Add func(data T, priority int, configs ...JobConfigFunc) EnqueuedJob[R]

+ AddAll func(items []PQItem[T]) EnqueuedGroupJob[R]

+ type Queue interface

+ Add func(data T, configs ...JobConfigFunc) EnqueuedJob[R]

+ AddAll func(data []Item[T]) EnqueuedGroupJob[R]

+ type Result struct

+ Data T

+ Err error

+ JobId string

+ type VoidWorkerFunc func(T)

+ type Worker interface

+ Copy func(config ...any) IWorkerBinder[T, R]

+ CurrentProcessingCount func() uint32

+ IsPaused func() bool

+ IsRunning func() bool

+ IsStopped func() bool

+ Pause func() Worker[T, R]

+ PauseAndWait func()

+ Restart func() error

+ Resume func() error

+ Status func() string

+ Stop func()

+ type WorkerErrFunc func(T) error

+ type WorkerFunc func(T) (R, error)

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL