varmq

README

A typed worker pool and message queue for Go. Each worker is its own pool and pulls jobs from any mix of sources: in-memory, priority, persistent and distributed queues, or topic subscriptions. Go generics keep every job type-safe.

Built for throughput: lock-free appends, batched dispatch and small in-memory jobs keep the hot path lean (benchmarks). Pause, stop, purge or resize a pool safely, inspect it over REST, and plug in your own storage through adapters.

Features · Quick Start · Highlighted Features · How It Works · Benchmarks · API Reference · Contributing

✨ Features

• ⚡ High performance: Optimized for high throughput with minimal overhead, even under heavy load. see benchmarks
• 🛠️ Queue types:
  • Standard queues for in-memory processing
  • Priority queues for importance-based ordering
  • Persistent queues that survive restarts, via adapters
  • Distributed queues shared across instances, via adapters
• 🧩 Worker abstractions:
  • NewWorker - Fire-and-forget operations (most performant)
  • NewErrWorker - Returns only error (when result isn't needed)
  • NewResultWorker - Returns result and error
• 🚦 Concurrency control: Fine-grained control over worker pool size, dynamic tuning and idle workers management
• 🧬 Multi Queue Binding: Bind multiple queues to a single worker
• 📢 Topic Fan-out: Publish once, every subscribed worker gets its own job from one shared log
• 📡 REST API: Built-in HTTP endpoints to inspect worker status and manage state (pause, resume, stop, restart)
• 🧩 Extensible: Build your own storage adapters by implementing VarMQ's internal queue interfaces

Quick Start

Installation

go get github.com/goptics/varmq

Basic Usage

package main

import (
    "fmt"
    "time"

    "github.com/goptics/varmq"
)

func main() {
  worker := varmq.NewWorker(func(j varmq.Job[int]) {
    fmt.Printf("Processing %d\n", j.Data())
    time.Sleep(1 * time.Second)
  }) // default concurrency is 1
  defer worker.WaitUntilIdle()
  queue := worker.BindQueue()

  queue.Add(1)
  queue.Add(2)
}

↗️ Run it on Playground

Priority Queue

You can use priority queue to prioritize jobs based on their priority. Lower number = higher priority

// just bind priority queue
queue := worker.BindPriorityQueue()

// add jobs to priority queue
for i := range 10 {
    queue.Add(i, i%2) // prioritize even tasks
}

↗️ Run it on Playground

Jobs waiting in a priority queue are reordered, so a higher-priority job added later is dispatched first:

Concurrency

worker := varmq.NewWorker(func(j varmq.Job[int]) {
  fmt.Printf("Processing %d\n", j.Data())
  time.Sleep(500 * time.Millisecond)
}, 10) // with concurrency 10 or set 0 for parallelism
defer worker.WaitUntilIdle()
queue := worker.BindQueue()

for i := range 100 {
  queue.Add(i)
}

↗️ Run it on Playground

[!Note] Concurrency is a limit, not a pool size. A worker starts with a single goroutine and spawns more only while jobs are waiting, up to the limit. As the queue drains, the extra goroutines exit one by one, keeping one idle (or the share set by varmq.WithMinIdleWorkerRatio). To keep idle goroutines around for reuse, set varmq.WithIdleWorkerExpiryDuration: they are then removed only after sitting idle that long.

How jobs reach workers, and why some arrive in batches: How It Works.

💡 Highlighted Features

Persistent and Distributed Queues

VarMQ supports both persistent and distributed queue processing through adapter interfaces:

• Persistent Queues: Store jobs durably so they survive program restarts
• Distributed Queues: Process jobs across multiple systems

Usage is simple:

// For persistent queues (with any IPersistentQueue adapter)
queue := worker.WithPersistentQueue(persistentQueueAdapter)

// For distributed queues (with any IDistributedQueue adapter)
queue := worker.WithDistributedQueue(distributedQueueAdapter)

See complete working examples in the examples directory:

• Persistent Queue Example (SQLite)
• Persistent Queue Example (Redis)
• Distributed Queue Example (Redis)

Create your own adapters by implementing the IPersistentQueue or IDistributedQueue interfaces.

[!Note] Before testing examples, make sure to start the Redis server using docker compose up -d.

Built-in adapters

• ⚡ Redis: redisq
• 🗃️ SQLite: sqliteq
• 🦆 DuckDB: duckq
• 🐘 PostgreSQL: 🔄 Upcoming

Multi Queue Binds

Bind multiple queues to a single worker, enabling efficient processing of jobs from different sources with configurable strategies. The worker supports four strategies:

1. Priority (default - prioritizes higher priority queues that have pending jobs)
2. RoundRobin (cycles through queues equally)
3. MaxLen (prioritizes queues with more jobs)
4. MinLen (prioritizes queues with fewer jobs)

worker := varmq.NewWorker(func(j varmq.Job[string]) {
  fmt.Println("Processing:", j.Data())
  time.Sleep(500 * time.Millisecond) // Simulate work
}) // change strategy through using varmq.WithStrategy default is varmq.Priority
defer worker.WaitUntilIdle()

// Bind to standard queues in chronological priority order
// You can change queue priority using varmq.WithQueuePriority function
q1 := worker.BindQueue()         // highest
q2 := worker.BindQueue()         // medium
pq := worker.BindPriorityQueue() // lowest

for i := range 15 {
  q2.Add(fmt.Sprintf("Task queue-2 %d", i))
}

for i := range 10 {
  pq.Add(fmt.Sprintf("Task priority-queue %d", i), i%2) // prioritize even tasks
}

for i := range 10 {
  q1.Add(fmt.Sprintf("Task queue-1 %d", i))
}

↗️ Run it on Playground

The Source Manager picks which bound queue the next job comes from. With varmq.WithStrategy(varmq.RoundRobin) it takes turns across the queues and skips any that are empty:

Topic (Fan-out)

A Topic is the opposite of multi queue binding: one producer, many workers. Publish once and every subscribed worker gets its own job. Publish returns false when no subscription is attached, the topic is closed, or it is full.

	Queue	Topic
Who gets a job	each job is processed once, by one worker	every subscribed worker gets its own job
Storage	the job is removed once a worker takes it	the payload is stored once; each worker reads it with its own cursor, and it's freed after every worker has read it
Producer	queue.Add(x)	topic.Publish(x)
Late joiner	sees jobs still waiting in the queue	sees only what's published after it subscribed

topic := varmq.NewTopic[int]()

email := varmq.NewWorker(func(j varmq.Job[int]) {
  fmt.Printf("Emailing order %d\n", j.Data())
  time.Sleep(1 * time.Second)
}) // concurrency 1
stock := varmq.NewWorker(func(j varmq.Job[int]) {
  fmt.Printf("Updating stock for order %d\n", j.Data())
  time.Sleep(300 * time.Millisecond)
}, 2) // concurrency 2
defer email.WaitUntilIdle()
defer stock.WaitUntilIdle()

email.WithTopic(topic)
stock.WithTopic(topic)

for order := range 4 {
  topic.Publish(order) // both workers get all 4 orders
}

↗️ Run it on Playground

Publish stores the payload once in the topic's log and wakes every subscribed worker. Each subscription reads the log through its own cursor, so the fast stock worker runs ahead without waiting for the email worker. A payload stays in the log until every cursor has read it, and a chunk of the log is freed once the slowest cursor has moved past it:

Result and Error Worker

VarMQ provides a NewResultWorker that returns both the result and error for each job processed. This is useful when you need to handle both success and failure cases.

worker := varmq.NewResultWorker(func(j varmq.Job[string]) (int, error) {
  fmt.Println("Processing:", j.Data())
  time.Sleep(500 * time.Millisecond) // Simulate work
  data := j.Data()

  if data == "error" {
    return 0, errors.New("error occurred")
  }

  return len(data), nil
})
defer worker.WaitUntilIdle()
queue := worker.BindQueue()

// Add returns at once; Result blocks until that job is done
if job, ok := queue.Add("The length of this string is 31"); ok {
  result, err := job.Result()
  fmt.Println("Result:", result, err) // Result: 31 <nil>
}

if job, ok := queue.Add("error"); ok {
  _, err := job.Result()
  fmt.Println("Error:", err) // Error: error occurred
}

↗️ Run it on Playground

NewErrWorker is similar to NewResultWorker but it returns only error.

Function Helpers

VarMQ provides helper functions that enable direct function submission similar to the Submit() pattern in other pool packages like Pond or Ants

• Func(): For basic functions with no return values - use with NewWorker
• ErrFunc(): For functions that return errors - use with NewErrWorker
• ResultFunc[R](): For functions that return a result and error - use with NewResultWorker

worker := varmq.NewWorker(varmq.Func(), 10)
defer worker.WaitUntilIdle()

queue := worker.BindQueue()

for i := range 100 {
    queue.Add(func() {
        time.Sleep(500 * time.Millisecond)
        fmt.Println("Processing", i)
    })
}

↗️ Run it on Playground

[!Important] Function helpers don't support persistence or distribution since functions cannot be serialized.

Lifecycle and Errors

A worker can be paused, resumed, resized and restarted while it runs. Each call returns at once; jobs already running always finish.

worker := varmq.NewWorker(process, 4,
  varmq.WithContext(ctx), // cancelling ctx stops the worker
  varmq.WithErrHandler(func(w varmq.Worker, err error) {
    log.Println("varmq:", err) // dequeue, parse and other internal errors
  }),
)

worker.Pause()     // stop taking new jobs; unstarted jobs stay queued
worker.Resume()    // carry on
worker.TunePool(8) // change concurrency on the fly
worker.Restart()   // stop and start again, keeping pending jobs
worker.Stop()      // graceful stop

worker.Wait()          // block until no job is running or pending
worker.WaitUntilIdle() // block until the worker is Idle (never returns while paused or stopped)

Status() reports the state (Running, Idle, Paused, Stopped, ...), and PauseAndWait, StopAndWait, WaitUntilPaused and WaitUntilStopped block until a state is reached.

REST API

VarMQ provides a built-in HTTP REST API for inspecting worker status and managing worker state remotely. Register the handler on any http.Server to expose endpoints:

mux := http.NewServeMux()
mux.Handle("/varmq/", varmq.Handler("/varmq"))
http.ListenAndServe(":8080", mux)

Available endpoints:

Method	Path	Description
GET	/health	Health check
GET	/workers	List all workers (summary)
GET	/workers/{name}	Get worker details + metrics
PATCH	/workers/{name}/actions/{action}	Pause, resume, stop, restart
PATCH	/workers/{name}/config/concurrency/{n}	Adjust concurrency

[!Warning] The REST API is unauthenticated by design. Protect it with middleware, network isolation, or a reverse proxy before exposing to untrusted networks.

How It Works

Event loop

A single event loop sleeps until something signals it: Add() putting a job in the queue, or a worker finishing one. On each signal it hands jobs to workers while the queue has work and a worker slot is free, then goes back to waiting.

Worker pool

Every Add (and every job in AddAll) tries to wake the event loop, but the signal channel holds only one pending signal, so the extra signals are simply dropped. One wake-up is enough: the loop keeps dispatching jobs back to back while the queue has work and a worker slot is free, and it hands each worker its share of the backlog in one go (a batch) rather than one job per wake-up.

Batched dispatch

Handing a job to a worker costs the event loop about the same whether the job takes a nanosecond or a minute. With many short jobs, handing them out one by one would keep the loop busier than the workers. So a worker can receive several jobs at once, a batch, and run them one after the other.

A batch holds the smallest of three numbers:

• Fair share. The waiting jobs split evenly across the workers, so every worker gets a job before any worker gets two.
• What the loop needs. VarMQ measures how long jobs take and gives each worker just enough that, all together, the workers come back for more about once every 32 µs. Until a job has been measured, a worker gets one at a time.
• 32, the hard cap.

With enough jobs waiting, that works out to:

Setup	Jobs per hand-off
10 workers, 3 s jobs	1, so jobs start in queue order
24 workers, 1 ms jobs	1
3,000 workers, 10 ms jobs	10
10,000+ workers, 10 ms jobs	32
Concurrency 1, any jobs	up to 32: one worker runs jobs in order anyway

So slow jobs go out one at a time and keep their order, and batches only grow when the loop would otherwise become the bottleneck. The job time comes from the last finished batch: if jobs suddenly get slower, one worker can still receive a batch sized for the old, fast jobs before the size adjusts.

Below, 8 cheap jobs with concurrency 2: the first worker takes 4, a second worker is started for 2, and the last 2 go out one at a time as workers free up.

While jobs wait in a batch. A worker runs its batch one job after another and tells the loop once, when the batch is done. Jobs still waiting their turn are Queued: NumPending counts them, NumProcessing does not, and you can still close them.

On Stop or Pause. Each worker finishes the job it is running and keeps the rest of its batch aside (parks them). Parked jobs stay Queued. NumPending counts them, but the queue's own Len does not, since they have already left the queue. On Restart or Resume they run first, in their original order. A Purge of their queue closes them at once.

Capacity. A queue's capacity limit counts only jobs still in the queue, not jobs in a batch or parked, so up to 32 more per worker can be pending.

Which queues batch.

Queue	Batches?	Why
In-memory queue, topic	Yes
Priority queue	No	A higher-priority job added later must not wait behind a batch.
Persistent or distributed queue	No	A worker must not hold unacknowledged messages that other consumers could take.
Worker bound to several queues	No	The strategy picks a queue for every single job.

Benchmarks

Comparison with Other Packages

VarMQ is benchmarked against ants, pond v2, conc, workerpool and two hand-rolled channel pools in the benchmark repository.

Feature matrix

Feature	VarMQ	ants	pond v2	conc	workerpool
Generic (type-safe) API	✅	✅	✅	✅	❌
Tasks return results and errors	✅	❌	✅	✅	❌
Task groups (submit many, wait once)	✅	❌	✅	✅	❌
Priority queue	✅	❌	❌	❌	❌
Persistent queues (survive restarts)	✅	❌	❌	❌	❌
Distributed queues (across instances)	✅	❌	❌	❌	❌
Several queues bound to one pool	✅	❌	❌	❌	❌
Topic fan-out (publish once, many pools)	✅	❌	❌	❌	❌
Bounded queue, reject when full	✅	✅	✅	❌	❌
Pause and resume	✅	❌	❌	❌	✅
Restart after stop	✅	✅	❌	❌	❌
Resize pool at runtime	✅	✅	✅	❌	❌
Cancel work via context	✅	❌	✅	✅	❌
Panic recovery	✅	✅	✅	❌¹	❌
Task counters (submitted, failed, ...)	✅	❌	✅	❌	❌
Built-in REST API	✅	❌	❌	❌	❌

_{Checked against each package's docs, October 2026. ¹ conc re-raises a task's panic from Wait by design.}

Performance

Every pool gets 50K workers and runs three kinds of task (sleep 10 ms, a short CPU loop, and a mix of both), 5M tasks of each, sent by anywhere from 1 to 1M producers. Charts for the sleep task:

4 CPUs

Open live chart

8 CPUs

Open live chart

24 CPUs

Open live chart

• 📊 Benchmark Repository - Complete benchmark suite
• 📈 Interactive Charts - Every workload, pool size and CPU count
• 📋 Bench Logs - Release benchmark tracking

Summary: time to run all 15M tasks per CPU count and memory per 1M tasks, lower is better; throughput in million tasks per second, averaged over every run, higher is better.

Pool	4 CPUs (s)	8 CPUs (s)	24 CPUs (s)	Throughput (M/s)	Memory (MB)
VarMQ	5.10	3.55	3.97	4.90	115
BufferedPool²	5.57	5.44	10.38	2.61	23
ants	6.71	6.53	8.11	2.22	16
GoroutinePool²	7.45	7.26	14.44	1.79	15
conc	8.66	8.65	16.75	1.56	15
workerpool	10.72	12.76	16.00	1.33	28
pond v2	21.82	23.20	46.76	0.57	317

Why more memory: Add never blocks, so VarMQ queues every task as a job (ID, status, result handle) until a worker takes it. ants, conc and the unbuffered pool make the producer wait instead, so far fewer tasks are held at once.

_{² Hand-rolled: fixed workers reading a channel, unbuffered (GoroutinePool) or with 1M slots (BufferedPool). Measured on an i7-13700.}

Micro-benchmarks

Cost of a single Add, AddAll and Publish call

goos: linux
goarch: amd64
pkg: github.com/goptics/varmq
cpu: 13th Gen Intel(R) Core(TM) i7-13700

Add

Command: go test -run=^$ -benchmem -bench '^(BenchmarkAdd)$' -cpu=1

Why use -cpu=1? Since the benchmark doesn’t test with more than 1 concurrent worker, a single CPU is ideal to accurately measure performance.

Worker Type	Queue Type	Time (ns/op)	Memory (B/op)	Allocations (allocs/op)
Worker	Queue	1,020	86	1
	Priority	1,057	64	1
ErrWorker	ErrQueue	1,055	110	1
	ErrPriority	1,093	88	1
ResultWorker	ResultQueue	1,053	102	1
	ResultPriority	1,093	80	1

AddAll

Command: go test -run=^$ -benchmem -bench '^(BenchmarkAddAll)$' -cpu=1

Worker Type	Queue Type	Time (ns/op)	Memory (B/op)	Allocations (allocs/op)
Worker	Queue	144,307	86,163	1,004
	Priority	700,103	64,129	1,002
ErrWorker	ErrQueue	171,741	126,701	1,507
	ErrPriority	745,010	104,669	1,505
ResultWorker	ResultQueue	181,409	143,299	1,007
	ResultPriority	735,915	121,268	1,005

[!Note]

AddAll benchmarks use a batch of 1000 items per call. The reported numbers (ns/op, B/op, allocs/op) are totals for the whole batch. For per-item values, divide each by 1000. e.g. for default Queue, the average time per item is approximately 144ns, against about 1,020ns for a single Add followed by Wait.

Why is AddAll so much faster per item than individual Add calls?

1. Batched dispatch: with a thousand jobs queued at once, the event loop hands each worker its share of the backlog in one wake-up instead of one job per wake-up, so the hand-off cost is spread over the batch. In-memory FIFO queues and topics do this; priority queues deliberately take one job per hand-off (a higher-priority job must never wait behind a batch), which is why their AddAll gains much less.
2. One wait, one group: a single group job tracks the whole batch and the caller waits once, instead of a full enqueue, wake-up, run, wake-the-caller round trip per job.

Publish

Command: go test -run=^$ -benchmem -bench '^(BenchmarkPublish)$' -cpu=1

Worker Type	Queue Type	Time (ns/op)	Memory (B/op)	Allocations (allocs/op)
Worker	Topic	126.8	76	1

One worker subscribed to the topic. Unlike the Add benchmark, it does not wait for each job: it publishes b.N payloads and waits once at the end, so the number is per-payload throughput with batched dispatch, not a round trip.

API Reference

For the complete package documentation, types, and method signatures, browse the GoDoc.

Star History

Contributing

Contributions are welcome! Please feel free to submit a Pull Request or open an issue.

Please note that this project has a Code of Conduct. By participating in this project, you agree to abide by its terms.

Documentation

Index

• Constants
• Variables
• func DefaultConcurrency(concurrency int)
• func DefaultCtx(ctx context.Context)
• func DefaultErrHandler(cb func(Worker, error))
• func DefaultIdleWorkerExpiryDuration(duration time.Duration)
• func DefaultJobIdGenerator(fn func() string)
• func DefaultMinIdleWorkerRatio(percentage uint8)
• func DefaultStoppedRetention(duration time.Duration)
• func DefaultStrategy(s Strategy)
• func ErrFunc() func(j Job[func() error]) error
• func Func() func(j Job[func()])
• func Handler(prefix string) http.Handler
• func NewServerMux() *http.ServeMux
• func ResultFunc[R any]() func(j Job[func() (R, error)]) (R, error)
• type Awaitable
• type CapacitySetter
• type ConfigFunc
  • func WithAutoRun(run bool) ConfigFunc
  • func WithConcurrency(concurrency int) ConfigFunc
  • func WithContext(ctx context.Context) ConfigFunc
  • func WithErrHandler(cb func(Worker, error)) ConfigFunc
  • func WithIdleWorkerExpiryDuration(duration time.Duration) ConfigFunc
  • func WithJobIdGenerator(fn func() string) ConfigFunc
  • func WithMinIdleWorkerRatio(percentage uint8) ConfigFunc
  • func WithName(name string) ConfigFunc
  • func WithStoppedRetention(duration time.Duration) ConfigFunc
  • func WithStrategy(s Strategy) ConfigFunc
• type DistributedPriorityQueue
  • func NewDistributedPriorityQueue[T any](internalQueue IDistributedPriorityQueue, configs ...QueueConfigFunc) DistributedPriorityQueue[T]
• type DistributedQueue
  • func NewDistributedQueue[T any](internalQueue IDistributedQueue, configs ...QueueConfigFunc) DistributedQueue[T]
• type Drainer
• type EnqueuedErrGroupJob
• type EnqueuedErrJob
• type EnqueuedGroupJob
• type EnqueuedJob
• type EnqueuedResultGroupJob
• type EnqueuedResultJob
• type ErrPriorityQueue
• type ErrQueue
• type IAcknowledgeable
• type IBaseQueue
• type IDistributedPriorityQueue
• type IDistributedQueue
• type IErrWorkerBinder
  • func NewErrWorker[T any](wf func(j Job[T]) error, config ...any) IErrWorkerBinder[T]
• type IExternalBaseQueue
• type IPersistentPriorityQueue
• type IPersistentQueue
• type IPriorityQueue
• type IQueue
• type IResultWorkerBinder
  • func NewResultWorker[T, R any](wf func(j Job[T]) (R, error), config ...any) IResultWorkerBinder[T, R]
• type ISubscribable
• type IWorkerBinder
  • func NewWorker[T any](wf func(j Job[T]), config ...any) IWorkerBinder[T]
• type Identifiable
• type Item
• type Job
• type JobConfigFunc
  • func WithJobId(id string) JobConfigFunc
• type Metrics
• type PendingTracker
• type PersistentPriorityQueue
• type PersistentQueue
• type PriorityQueue
• type Queue
• type QueueConfigFunc
  • func WithQueueCapacity(capacity int) QueueConfigFunc
  • func WithQueuePriority(priority int) QueueConfigFunc
• type Result
• type ResultPriorityQueue
• type ResultQueue
• type StatusProvider
• type Strategy
• type Subscription
• type Topic
  • func NewTopic[T any](configs ...TopicConfigFunc) Topic[T]
• type TopicConfigFunc
  • func WithTopicCapacity(capacity int) TopicConfigFunc
  • func WithTopicPriority(priority int) TopicConfigFunc
• type Worker

Constants

const InstanceHeader = "X-Varmq-Instance"

InstanceHeader is the response header carrying InstanceID. Clients use it to tell which process answered when several instances sit behind one URL (e.g. a load balancer).

Variables

var (
	ErrJobProcessing    = errors.New("job is processing, you can't close processing job")
	ErrJobAlreadyClosed = errors.New("job is already closed")
	ErrAcknowledgeJob   = errors.New("queue failed to acknowledge job")
)

var (
	ErrRunningWorker    = errors.New("worker is already running")
	ErrNotRunningWorker = errors.New("worker is not running")
	ErrWorkerPaused     = errors.New("worker is paused")
	ErrWorkerStopped    = errors.New("worker is stopped")
	ErrWorkerPausing    = errors.New("worker is pausing")
	ErrWorkerStopping   = errors.New("worker is stopping")
	ErrSameConcurrency  = errors.New("worker already has the same concurrency")
	ErrFailedToDequeue  = errors.New("failed to dequeue job")
	ErrFailedToCastJob  = errors.New("failed to cast job")
	ErrParseJob         = errors.New("failed to parse job")
)

var InstanceID, _ = os.Hostname()

InstanceID identifies this process in every REST API response. It defaults to the hostname (the pod name on Kubernetes), or "" if that is unavailable; set it before serving when several instances share a host.

var WorkerRegistry sync.Map

WorkerRegistry safely stores reference to all active workers by Name

Functions

func DefaultConcurrency

func DefaultConcurrency(concurrency int)

DefaultConcurrency sets the default concurrency level for all newly created workers. If concurrency is less than 1, it defaults to the number of CPU cores.

Parameters:

• concurrency: The number of concurrent workers to use. A value less than 1 will default to the number of CPU cores.

Default: 1

func DefaultCtx

func DefaultCtx(ctx context.Context)

DefaultCtx sets the default context for all newly created workers. The context can be used to stop all worker goroutines when cancelled.

Parameters:

• ctx: The context to use

Default: nil (no context)

func DefaultErrHandler

func DefaultErrHandler(cb func(Worker, error))

DefaultErrHandler sets the default error handler callback for all newly created workers. The provided function is called whenever a worker encounters an error, receiving the worker instance and the error as arguments. It runs synchronously on the goroutine that reports the error and must not block. The handler is invoked for every error, including when the buffered Errs() channel is already full. If cb is nil, the call is a no-op.

Parameters:

• cb: A function receiving the worker and the error.

Default: A no-op handler (errors are only sent to the Errs() channel).

func DefaultIdleWorkerExpiryDuration

func DefaultIdleWorkerExpiryDuration(duration time.Duration)

DefaultIdleWorkerExpiryDuration sets the default time period after which idle workers are automatically removed from the worker pool for all newly created workers.

This helps optimize resource usage by removing unnecessary idle workers when the system experiences prolonged periods of low activity.

Parameters:

• duration: The time period a worker can remain idle before being removed (e.g., 30*time.Second, 5*time.Minute)

Default: Not set — exactly one idle worker is always maintained in the pool.

func DefaultJobIdGenerator

func DefaultJobIdGenerator(fn func() string)

DefaultJobIdGenerator sets the default job ID generator function for all newly created workers. The provided function will be called to generate a unique ID for each job. If fn is nil, the call is a no-op.

Parameters:

• fn: A function that returns a unique string ID.

Default: A no-op generator that returns an empty string (no job ID).

func DefaultMinIdleWorkerRatio

func DefaultMinIdleWorkerRatio(percentage uint8)

DefaultMinIdleWorkerRatio sets the default minimum percentage of idle workers to maintain in the pool as a proportion of the total concurrency level for all newly created workers.

Maintaining some idle workers allows the system to respond quickly to incoming jobs without the overhead of creating new worker goroutines.

Parameters:

• percentage: An integer between 1-100 representing the percentage of workers to keep idle. Values outside the range 1-100 are automatically clamped (0 becomes 1, >100 becomes 100).

Default: At least one idle worker is always maintained in the pool.

func DefaultStoppedRetention

func DefaultStoppedRetention(duration time.Duration)

DefaultStoppedRetention sets the default time after a stopped worker is automatically removed from the global registry.

When a worker with a name is stopped, a timer starts for this duration. If the timer expires before the worker is started again, the worker is removed from WorkerRegistry.

Set to 0 to disable (worker stays in registry forever after stop).

Default: 1 minute

func DefaultStrategy

func DefaultStrategy(s Strategy)

DefaultStrategy sets the default queue selection strategy for all newly created workers.

Available strategies are:

• Priority: Selects the queue with the highest priority (default)
• RoundRobin: Selects queues in a round-robin fashion
• MaxLen: Selects the queue with the most items
• MinLen: Selects the queue with the fewest items

Parameters:

• s: The strategy to use

Default: Priority

func ErrFunc

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

ErrFunc is a helper function for functions that return errors. It enables direct submission of functions that return error values.

Usage:

queue := varmq.NewErrWorker(varmq.ErrFunc()).BindQueue()
job, ok := queue.Add(func() error {
    if err := doSomething(); err != nil {
        return fmt.Errorf("failed to do something: %w", err)
    }
    return nil
})

if err := job.Err(); err != nil {
    log.Printf("Job failed: %v", err)
}

func Func

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

Func is a helper function that enables direct function submission to VarMQ workers. Instead of creating custom data types and worker functions, you can pass functions directly.

Usage:

queue := varmq.NewWorker(varmq.Func()).BindQueue()
queue.Add(func() {
    fmt.Println("Hello from worker!")
})

Note: Func does not support persistence or distribution since functions are not serializable. For persistent or distributed job queues, use custom or primitive data types that can be serialized to JSON instead of passing functions directly.

func Handler

func Handler(prefix string) http.Handler

Handler returns an http.Handler that serves the varmq REST API under the given prefix. This is a convenience wrapper around NewServerMux.

Example:

mainMux := http.NewServeMux()
mainMux.Handle("/varmq/", varmq.Handler("/varmq"))
http.ListenAndServe(":8080", mainMux)

Handler sets InstanceHeader on every response, including the 404 and 405 replies ServeMux generates itself for unknown paths or methods.

func NewServerMux

func NewServerMux() *http.ServeMux

NewServerMux returns a new http.ServeMux with all the REST API routes configured. The returned mux has no path prefix, so it can be used directly or mounted under any prefix using Handler.

func ResultFunc

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

ResultFunc is a generic helper function for functions that return both a result and an error. It enables direct submission of functions with return values of any type.

Usage:

queue := varmq.NewResultWorker(varmq.ResultFunc[string]()).BindQueue()
job, ok := queue.Add(func() (string, error) {
    result, err := fetchData()
    if err != nil {
        return "", fmt.Errorf("failed to fetch: %w", err)
    }
    return result, nil
})

result, err := job.Result()
if err != nil {
    log.Printf("Job failed: %v", err)
} else {
    fmt.Printf("Result: %s", result)
}

The generic type parameter R specifies the return type of your function.

Types

type Awaitable

type Awaitable interface {
	// Wait blocks the caller until the job has completed processing.
	// This method can be used to synchronize execution flow with job completion.
	// If the job has already finished, this method returns immediately.
	Wait()
}

type CapacitySetter

type CapacitySetter interface {
	SetCapacity(cap int)
}

CapacitySetter is the interface for setting the capacity of a queue. (Optional)

type ConfigFunc

type ConfigFunc func(*configs)

ConfigFunc is a function that configures a worker.

func WithAutoRun

func WithAutoRun(run bool) ConfigFunc

WithAutoRun configures whether the worker automatically starts when created. When autoRun is true (the default), the worker begins processing immediately after creation. When set to false, the worker remains in an initiated state until Start() is called manually.

Default: true

func WithConcurrency

func WithConcurrency(concurrency int) ConfigFunc

WithConcurrency sets the concurrency level for the worker. If not set, the default concurrency level is 1. If concurrency is less than 1, it defaults to number of CPU cores.

func WithContext

func WithContext(ctx context.Context) ConfigFunc

WithContext configures the context for the worker.

Each worker is associated with a context that is used to stop all goroutines when the worker is stopped. You can create a custom context and pass it to the worker to stop all workers when the context is cancelled.

Parameters:

• ctx: The context to use

Default: If this option is not set, the default is nil context.

func WithErrHandler

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

WithErrHandler sets the error handler callback for the worker. The provided function is called whenever the worker encounters an error, receiving the worker instance and the error as arguments. It runs synchronously on the goroutine that reports the error and must not block. The handler is invoked for every error, including when the buffered Errs() channel is already full. If cb is nil, the call is a no-op.

Parameters:

• cb: A function receiving the worker and the error.

func WithIdleWorkerExpiryDuration

func WithIdleWorkerExpiryDuration(duration time.Duration) ConfigFunc

WithIdleWorkerExpiryDuration configures the time period after which idle workers are automatically removed from the worker pool to conserve system resources.

This setting helps optimize resource usage by removing unnecessary idle workers when the system experiences prolonged periods of low activity. When job volume increases again, new workers will be created as needed up to the configured concurrency level.

Parameters:

• duration: The time period a worker can remain idle before being removed (e.g., 30*time.Second, 5*time.Minute)

Default behavior: If this option is not set, exactly one idle worker will always be maintained in the pool, regardless of how high the concurrency level is configured.

When job load decreases below the concurrency level, the system will immediately begin removing excess idle workers according to this expiry duration setting.

func WithJobIdGenerator

func WithJobIdGenerator(fn func() string) ConfigFunc

WithJobIdGenerator sets the job ID generator function for the worker. If not set there wouldn't be any job id

func WithMinIdleWorkerRatio

func WithMinIdleWorkerRatio(percentage uint8) ConfigFunc

WithMinIdleWorkerRatio configures the minimum percentage of idle workers to maintain in the pool as a proportion of the total concurrency level.

This configuration helps optimize resource usage by dynamically scaling the idle worker pool when the concurrency level changes. Maintaining some idle workers allows the system to respond quickly to incoming jobs without the overhead of creating new worker goroutines.

Parameters:

• percentage: An integer between 1-100 representing the percentage of workers to keep idle

Examples:

• WithMinIdleWorkerRatio(20): With concurrency=10, maintains 2 idle workers (20%)
• WithMinIdleWorkerRatio(50): With concurrency=10, maintains 5 idle workers (50%)

Values outside the range 1-100 are automatically clamped (0 becomes 1, >100 becomes 100). By default there is always at least one idle worker inside the pool.

func WithName

func WithName(name string) ConfigFunc

WithName configures the name for the worker.

Providing a non-empty name sets the worker name, which is used for registration in WorkerRegistry when the worker starts. If left empty, no default name is generated and the worker will not be registered or queryable via the Mux HTTP API.

func WithStoppedRetention

func WithStoppedRetention(duration time.Duration) ConfigFunc

WithStoppedRetention configures the time after a stopped worker is automatically removed from the global registry.

When the worker stops, a timer starts for this duration. If the timer expires before the worker is started again, the worker is removed from WorkerRegistry.

Set to 0 to disable (worker stays in registry forever after stop).

Default: 1 minute (from DefaultStoppedRetention)

func WithStrategy

func WithStrategy(s Strategy) ConfigFunc

WithStrategy configures the queue selection strategy for the worker.

The worker uses this strategy to determine which queue to pull jobs from when multiple queues are registered. Available strategies are:

• Priority: Selects the queue with the highest priority (default)
• RoundRobin: Selects queues in a round-robin fashion
• MaxLen: Selects the queue with the most items
• MinLen: Selects the queue with the fewest items

Parameters:

• strategy: The strategy to use (Priority, RoundRobin, MaxLen or MinLen)

Default: If this option is not set, Priority strategy will be used.

type DistributedPriorityQueue

type DistributedPriorityQueue[T any] interface {
	IExternalBaseQueue

	// Add data to the queue with priority
	Add(data T, priority int, configs ...JobConfigFunc) bool
}

DistributedPriorityQueue is a external queue wrapper of an any IDistributedPriorityQueue

func NewDistributedPriorityQueue

func NewDistributedPriorityQueue[T any](internalQueue IDistributedPriorityQueue, configs ...QueueConfigFunc) DistributedPriorityQueue[T]

type DistributedQueue

type DistributedQueue[T any] interface {
	IExternalBaseQueue
	// Add data to the queue
	Add(data T, configs ...JobConfigFunc) bool
}

DistributedQueue is a external queue wrapper of an any IDistributedQueue

func NewDistributedQueue

func NewDistributedQueue[T any](internalQueue IDistributedQueue, configs ...QueueConfigFunc) DistributedQueue[T]

type Drainer

type Drainer interface {
	// Drain discards the results or errors a group job's channel still holds,
	// so a caller that does not read them does not keep the channel full.
	// A single job keeps its result in place and has no channel: Drain is a
	// no-op there, and Err or Result still return the value.
	Drain()
}

type EnqueuedErrGroupJob

type EnqueuedErrGroupJob interface {
	// Errs returns a channel from which errors of the group jobs can be received.
	// It blocks until all jobs in the group are completed.
	Errs() <-chan error
	PendingTracker
	Awaitable
	Drainer
}

EnqueuedErrGroupJob is an interface representing an error group job that has been successfully enqueued. It extends PendingTracker, Awaitable, and Drainer interfaces to provide tracking, waiting, and cleanup capabilities. This interface is returned by the AddAll method when used with NewErrWorker, allowing clients to collect errors generated during job processing through the Errs() method

type EnqueuedErrJob

type EnqueuedErrJob interface {
	EnqueuedJob
	Drainer
	// Err retrieves any error that occurred during job execution.
	// This method blocks the caller until the job processing is complete.
	// If the job has already finished, it returns immediately with the error result.
	// A nil return value indicates successful job execution without errors.
	Err() error
}

EnqueuedErrJob is an interface representing an error job that has been successfully enqueued. It extends EnqueuedJob and Drainer interfaces to provide identification, status tracking, waiting, and cleanup capabilities. This interface is returned by the Add method when used with NewErrWorker, allowing clients to retrieve any error through the Err() method.

type EnqueuedGroupJob

type EnqueuedGroupJob interface {
	PendingTracker
	Awaitable
}

EnqueuedGroupJob is an interface representing a group job that has been successfully enqueued. It extends PendingTracker and Awaitable This interface is returned by the AddAll method when used with NewWorker

type EnqueuedJob

type EnqueuedJob interface {
	Identifiable
	StatusProvider
	Awaitable
	// Close marks the job as closed without removing it from the queue.
	// When a closed job is dequeued, the worker will skip processing it.
	// This operation takes constant time as it only updates the job's state.
	Close() error
}

EnqueuedJob is an interface representing a job that has been successfully enqueued. It extends Identifiable, StatusProvider, and Awaitable interfaces to provide identification, status tracking, and waiting capabilities. This interface is returned by the Add method when used with NewWorker

type EnqueuedResultGroupJob

type EnqueuedResultGroupJob[R any] interface {
	// Results returns a channel from which results of the group jobs can be received.
	// It blocks until all jobs in the group are completed.
	Results() <-chan Result[R]
	PendingTracker
	Awaitable
	Drainer
}

EnqueuedResultGroupJob is an interface representing a result group job that has been successfully enqueued. It extends PendingTracker, Awaitable, and Drainer interfaces to provide tracking, waiting, and cleanup capabilities. This interface is returned by the AddAll method when used with NewResultWorker, allowing clients to collect job results through the Results() method

type EnqueuedResultJob

type EnqueuedResultJob[R any] interface {
	EnqueuedJob
	Drainer
	// Result retrieves the computed result value and any error from the job execution.
	// This method blocks the caller until the job processing is complete.
	// If the job has already finished, it returns immediately with the result.
	Result() (R, error)
}

EnqueuedResultJob is an interface representing a result job that has been successfully enqueued. It extends EnqueuedJob and Drainer interfaces to provide identification, status tracking, waiting, and cleanup capabilities. This interface is returned by the Add method when used with NewResultWorker, allowing clients to retrieve the job result through the Result() method.

type ErrPriorityQueue

type ErrPriorityQueue[T any] interface {
	IExternalBaseQueue
	// Worker returns the bound worker.
	Worker() Worker
	// Add adds a new Job with the given priority to the queue and returns a EnqueuedErrJob to handle the job with error receiving.
	// Time complexity: O(log n)
	Add(data T, priority int, configs ...JobConfigFunc) (EnqueuedErrJob, bool)
	// AddAll adds multiple Jobs with the given priority to the queue and returns a EnqueuedErrGroupJob to handle the job with error receiving.
	// Time complexity: O(n log n) where n is the number of Jobs added
	AddAll(data []Item[T]) EnqueuedErrGroupJob
}

type ErrQueue

type ErrQueue[T any] interface {
	IExternalBaseQueue

	// Worker returns the bound worker.
	Worker() Worker
	// Add adds a new Job to the queue and returns a EnqueuedErrJob to handle the job with error receiving.
	// Time complexity: O(1)
	Add(data T, configs ...JobConfigFunc) (EnqueuedErrJob, bool)
	// AddAll adds multiple Jobs to the queue and returns a EnqueuedErrGroupJob to handle the job with error receiving.
	// Time complexity: O(n) where n is the number of Jobs added
	AddAll(data []Item[T]) EnqueuedErrGroupJob
}

type IAcknowledgeable

type IAcknowledgeable interface {
	// Returns true if the item was successfully acknowledged, false otherwise.
	Acknowledge(ackID string) bool
	// DequeueWithAckId dequeues an item from the queue
	// Returns the item, a boolean indicating if the operation was successful, and the acknowledgment ID.
	DequeueWithAckId() (any, bool, string)
}

IAcknowledgeable is the root interface of acknowledgeable operations.

type IBaseQueue

type IBaseQueue interface {
	Len() int
	Dequeue() (any, bool)
	Values() []any
	Purge()
	Close() error
}

IBaseQueue is the root interface of queue operations. workers queue needs to implement this interface.

type IDistributedPriorityQueue

type IDistributedPriorityQueue interface {
	IPersistentPriorityQueue
	ISubscribable
}

IDistributedPriorityQueue is the root interface of distributed priority queue operations.

type IDistributedQueue

type IDistributedQueue interface {
	IPersistentQueue
	ISubscribable
}

IDistributedQueue is the root interface of distributed queue operations.

type IErrWorkerBinder

type IErrWorkerBinder[T any] interface {
	Worker

	// BindQueue binds the worker to a standard Queue implementation.
	// It creates a new Queue with default settings and connects with the worker.
	//
	// Parameters:
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - ErrQueue[T]: A fully configured ErrQueue that automatically processes jobs using this worker.
	//
	// Example usage:
	//   errQueue := errWorker.BindQueue()
	//   errJob, ok := errQueue.Add(data)  // Enqueues a job that will be processed by the error worker
	BindQueue(configs ...QueueConfigFunc) ErrQueue[T]
	// WithQueue binds the worker to a custom Queue implementation.
	// This method allows you to use your own queue implementation as long as it
	// satisfies the IQueue interface. This is useful when you need specialized
	// queueing behavior beyond what the standard queue provides.
	//
	// Parameters:
	//   - q IQueue: A custom queue implementation that satisfies the IQueue interface.
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - ErrQueue[T]: A fully configured ErrQueue that uses the provided implementation and processes jobs with this worker.
	//
	// Example usage:
	//   customQueue := NewCustomQueue()
	//   errQueue := errWorker.WithQueue(customQueue)
	//   errJob, ok := errQueue.Add(data)  // Enqueues a job that will be processed by the error worker
	WithQueue(q IQueue, configs ...QueueConfigFunc) ErrQueue[T]

	// BindPriorityQueue binds the worker to a standard PriorityQueue implementation.
	// It creates a new PriorityQueue with default settings and connects with the worker.
	// Use this when you need to process jobs based on priority rather than FIFO order.
	//
	// Returns:
	//   - ErrPriorityQueue[T]: A fully configured ErrPriorityQueue that automatically processes jobs using this worker.
	//
	// Example usage:
	//   errPriorityQueue := errWorker.BindPriorityQueue()
	//   errJob, ok := errPriorityQueue.Add(data, 5)  // Enqueues a job with priority 5 that will be processed by the error worker
	BindPriorityQueue(configs ...QueueConfigFunc) ErrPriorityQueue[T]

	// WithPriorityQueue binds the worker to a custom PriorityQueue implementation.
	// This method allows you to use your own priority queue implementation as long as it
	// satisfies the IPriorityQueue interface. This is useful when you need specialized
	// priority-based queueing beyond what the standard implementation provides.
	//
	// Parameters:
	//   - pq IPriorityQueue: A custom priority queue implementation that satisfies the IPriorityQueue interface.
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - ErrPriorityQueue[T]: A fully configured ErrPriorityQueue that uses the provided implementation and processes jobs with this worker.
	//
	// Example usage:
	//   customPriorityQueue := NewCustomPriorityQueue()
	//   errPriorityQueue := errWorker.WithPriorityQueue(customPriorityQueue)
	//   errJob, ok := errPriorityQueue.Add(data, 5)  // Enqueues a job with priority 5 that will be processed by the error worker
	WithPriorityQueue(pq IPriorityQueue, configs ...QueueConfigFunc) ErrPriorityQueue[T]
}

IErrWorkerBinder is the interface for binding error workers to different queue types. It provides methods to connect error workers with various queue implementations, enabling with queue binding capabilities for handling tasks of type T and collecting errors during processing.

func NewErrWorker

func NewErrWorker[T any](wf func(j Job[T]) error, config ...any) IErrWorkerBinder[T]

NewErrWorker creates a worker for operations that only return errors (no result value). This is useful for operations where you only care about success/failure status. If concurrency in config is not provided, it defaults to 1. and if concurrency is less than 1, it defaults to number of CPU cores. Unlike NewWorker, it can't be bound to distributed or persistent queue. Parameters:

• wf: Worker function that processes items and returns only error
• config: Optional configuration parameters (concurrency, idle worker ratio, etc.)

Example:

worker := NewErrWorker(func(data int) error {
    log.Printf("Processing: %d", data)
    return nil
}, 4) // 4 concurrent workers
queue := worker.BindQueue() // Bind to standard queue
priorityQueue := worker.BindPriorityQueue() // Bind to priority queue

type IExternalBaseQueue

type IExternalBaseQueue interface {
	// Deprecated: Use Queue's Len() method instead for a more idiomatic approach.
	// This method will be removed in an upcoming release.
	NumPending() int

	// Len returns the number of jobs in the queue. Jobs a worker has already
	// taken into a batch but not started are not in the queue; the worker's
	// NumPending counts them.
	Len() int
	// IsFull returns true if the queue has reached its configured capacity.
	// If no capacity is set, it always returns false. Capacity bounds the
	// jobs in storage: jobs a worker took in a batch, or parked on Stop or
	// Pause, are outside it, at most 32 per worker.
	IsFull() bool
	// Purge removes all pending Jobs from the queue.
	Purge()
	// Close closes the queue, no more jobs can be added to the queue after closing the queue.
	Close() error
}

type IPersistentPriorityQueue

type IPersistentPriorityQueue interface {
	IPriorityQueue
	IAcknowledgeable
}

IPersistentPriorityQueue is the root interface of persistent priority queue operations.

type IPersistentQueue

type IPersistentQueue interface {
	IQueue
	IAcknowledgeable
}

type IPriorityQueue

type IPriorityQueue interface {
	IBaseQueue
	Enqueue(item any, priority int) bool
}

IPriorityQueue is the root interface of priority queue operations.

type IQueue

type IQueue interface {
	IBaseQueue
	Enqueue(item any) bool
}

IQueue is the root interface of queue operations.

type IResultWorkerBinder

type IResultWorkerBinder[T, R any] interface {
	Worker

	// BindQueue binds the worker to a standard Queue implementation.
	// It creates a new Queue with default settings and connects the worker to it.
	// This is the simplest way to get a standard FIFO queue working with this worker.
	//
	// Parameters:
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - ResultQueue[T, R]: A fully configured Result Queue that automatically processes jobs using this worker.
	//
	// Example usage:
	//   queue := worker.BindQueue()
	//   queue.Add(data) // Enqueues a job that will be processed by the worker
	BindQueue(configs ...QueueConfigFunc) ResultQueue[T, R]

	// WithQueue binds the worker to a custom Queue implementation.
	// This method allows you to use your own queue implementation as long as it
	// satisfies the IQueue interface. This is useful when you need specialized
	// queueing behavior beyond what the standard queue provides.
	//
	// Parameters:
	//   - q IQueue: A custom queue implementation that satisfies the IQueue interface.
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - ResultQueue[T, R]: A Result Queue that uses the provided implementation and processes jobs with this worker.
	//
	// Example usage:
	//   customQueue := NewCustomQueue() // satisfies IQueue interface
	//   queue := worker.WithQueue(customQueue)
	WithQueue(q IQueue, configs ...QueueConfigFunc) ResultQueue[T, R]

	// BindPriorityQueue binds the worker to a standard PriorityQueue implementation.
	// It creates a new PriorityQueue with default settings and connects the worker to it.
	// Use this when you need to process jobs based on priority rather than FIFO order.
	//
	// Parameters:
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - ResultPriorityQueue[T, R]: A fully configured ResultPriorityQueue that processes jobs using this worker.
	//
	// Example usage:
	//   priorityQueue := worker.BindPriorityQueue() // satisfies IPriorityQueue interface
	//   priorityQueue.Add(data, 5) // Enqueue with priority 5
	BindPriorityQueue(configs ...QueueConfigFunc) ResultPriorityQueue[T, R]

	// WithPriorityQueue binds the worker to a custom PriorityQueue implementation.
	// This method allows you to use your own priority queue implementation as long as it
	// satisfies the IPriorityQueue interface. This is useful when you need specialized
	// priority-based queueing beyond what the standard implementation provides.
	//
	// Parameters:
	//   - pq IPriorityQueue: A custom priority queue implementation that satisfies the IPriorityQueue interface.
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - ResultPriorityQueue[T, R]: A Result PriorityQueue that uses the provided implementation and processes jobs with this worker.
	//
	// Example usage:
	//   customPriorityQueue := NewCustomPriorityQueue()
	//   priorityQueue := worker.WithPriorityQueue(customPriorityQueue)
	WithPriorityQueue(pq IPriorityQueue, configs ...QueueConfigFunc) ResultPriorityQueue[T, R]
}

IResultWorkerBinder is the base interface for binding workers to different queue types. It provides methods to connect workers with various queue implementations, enabling flexible worker-queue configurations. With queue binding capabilities for handling tasks of type T and producing results of type R.

func NewResultWorker

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

NewResultWorker creates a worker for operations that return a result value and error. If concurrency in config is not provided, it defaults to 1. and if concurrency is less than 1, it defaults to number of CPU cores. Unlike NewWorker, it can't be bound to distributed or persistent queue. Parameters:

• wf: Worker function that processes items and returns a result and error
• config: Optional configuration parameters (concurrency, idle worker ratio, etc.)

Example:

worker := NewResultWorker(func(data string) (int, error) {
    return len(data), nil
}, 4) // 4 concurrent workers
queue := worker.BindQueue() // Bind to standard queue
priorityQueue := worker.BindPriorityQueue() // Bind to priority queue

type ISubscribable

type ISubscribable interface {
	Subscribe(func(action string))
}

ISubscribable is the root interface of subscribable operations.

type IWorkerBinder

type IWorkerBinder[T any] interface {
	Worker
	// BindQueue binds the worker to a standard Queue implementation.
	// It creates a new Queue with default settings and connects the worker to it.
	// This is the simplest way to get a standard FIFO queue working with this worker.
	//
	// Parameters:
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - Queue[T]: A fully configured Queue that automatically processes jobs using this worker.
	//
	// Example usage:
	//   queue := worker.BindQueue()
	//   queue.Add(data) // Enqueues a job that will be processed by the worker
	BindQueue(configs ...QueueConfigFunc) Queue[T]

	// WithQueue binds the worker to a custom Queue implementation.
	// This method allows you to use your own queue implementation as long as it
	// satisfies the IQueue interface. This is useful when you need specialized
	// queueing behavior beyond what the standard queue provides.
	//
	// Parameters:
	//   - q IQueue: A custom queue implementation that satisfies the IQueue interface.
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - Queue[T]: A Queue that uses the provided implementation and processes jobs with this worker.
	//
	// Example usage:
	//   customQueue := NewCustomQueue() // satisfies IQueue interface
	//   queue := worker.WithQueue(customQueue)
	WithQueue(q IQueue, configs ...QueueConfigFunc) Queue[T]

	// BindPriorityQueue binds the worker to a standard PriorityQueue implementation.
	// It creates a new PriorityQueue with default settings and connects the worker to it.
	// Use this when you need to process jobs based on priority rather than FIFO order.
	//
	// Parameters:
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - PriorityQueue[T]: A fully configured PriorityQueue that processes jobs using this worker.
	//
	// Example usage:
	//   priorityQueue := worker.BindPriorityQueue() // satisfies IPriorityQueue interface
	//   priorityQueue.Add(data, 5) // Enqueue with priority 5
	BindPriorityQueue(configs ...QueueConfigFunc) PriorityQueue[T]

	// WithPriorityQueue binds the worker to a custom PriorityQueue implementation.
	// This method allows you to use your own priority queue implementation as long as it
	// satisfies the IPriorityQueue interface. This is useful when you need specialized
	// priority-based queueing beyond what the standard implementation provides.
	//
	// Parameters:
	//   - pq IPriorityQueue: A custom priority queue implementation that satisfies the IPriorityQueue interface.
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - PriorityQueue[T]: A PriorityQueue that uses the provided implementation and processes jobs with this worker.
	//
	// Example usage:
	//   customPriorityQueue := NewCustomPriorityQueue()
	//   priorityQueue := worker.WithPriorityQueue(customPriorityQueue)
	WithPriorityQueue(pq IPriorityQueue, configs ...QueueConfigFunc) PriorityQueue[T]

	// WithPersistentQueue binds the worker to a PersistentQueue.
	// PersistentQueue provides durability guarantees for jobs, ensuring they are not lost
	// even in the event of application crashes or restarts. This is useful for critical
	// workloads where job completion must be guaranteed.
	//
	// Parameters:
	//   - pq IPersistentQueue: A queue implementation that provides persistence capabilities.
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - PersistentQueue[T, any]: A persistent queue that ensures jobs are not lost and processes them with the worker.
	//
	// Example usage:
	//   persistentQueue := worker.WithPersistentQueue(redisQueue)
	WithPersistentQueue(pq IPersistentQueue, configs ...QueueConfigFunc) PersistentQueue[T]

	// WithPersistentPriorityQueue binds the worker to a PersistentPriorityQueue.
	// This combines the benefits of persistence with priority-based processing. Jobs are
	// both durable and processed according to their priority values.
	//
	// Parameters:
	//   - pq IPersistentPriorityQueue: A priority queue implementation that provides persistence capabilities.
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - PersistentPriorityQueue[T, any]: A persistent priority queue that processes jobs based on priority
	//     while ensuring they are not lost, using this worker for processing.
	//
	// Example usage:
	//   persistentPriorityQueue := worker.WithPersistentPriorityQueue(persistentPriorityQueue)
	WithPersistentPriorityQueue(pq IPersistentPriorityQueue, configs ...QueueConfigFunc) PersistentPriorityQueue[T]

	// WithDistributedQueue binds the worker to a DistributedQueue implementation.
	// Distributed queues allow job processing to be spread across multiple instances or processes,
	// enabling horizontal scaling of workers. Since void workers don't need to return results,
	// they are perfect for distributed processing scenarios.
	//
	// Parameters:
	//   - dq IDistributedQueue: A distributed queue implementation that satisfies the IDistributedQueue interface.
	//     This could be backed by Redis or any other distributed messaging system.
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - DistributedQueue[T, any]: A distributed queue that can distribute jobs across multiple workers or instances.
	//
	// Example usage:
	//   dq := NewDistributedQueue() // satisfies IDistributedQueue interface, might be backed by Redis or other systems
	//   distributedQueue := voidWorker.WithDistributedQueue(dq)
	//   distributedQueue.Add(data) // This job can be processed by any worker instance listening to this queue
	WithDistributedQueue(dq IDistributedQueue, configs ...QueueConfigFunc) DistributedQueue[T]

	// WithDistributedPriorityQueue binds the worker to a DistributedPriorityQueue implementation.
	// This combines distributed processing with priority-based job ordering. Jobs are distributed
	// across multiple instances but processed according to priority within each instance.
	// This is ideal for scenarios where some jobs are more urgent than others, but still need
	// to be processed in a distributed manner.
	//
	// Parameters:
	//   - dq IDistributedPriorityQueue: A distributed priority queue implementation that satisfies
	//     the IDistributedPriorityQueue interface. This could be backed by Redis or other
	//     distributed messaging systems that support priority-based message ordering.
	//   - configs ...QueueConfigFunc: Optional configuration functions to customize the queue behavior.
	//
	// Returns:
	//   - DistributedPriorityQueue[T, any]: A distributed priority queue that can distribute jobs
	//     across multiple workers while maintaining priority ordering.
	//
	// Example usage:
	//   priorityQueue := NewDistributedPriorityQueue() // satisfies IDistributedPriorityQueue interface, might be backed by Redis or other systems
	//   distributedPriorityQueue := voidWorker.WithDistributedPriorityQueue(priorityQueue)
	//   distributedPriorityQueue.Add(data, -1) // This job will be processed with higher priority
	WithDistributedPriorityQueue(dq IDistributedPriorityQueue, configs ...QueueConfigFunc) DistributedPriorityQueue[T]

	// WithTopic subscribes the worker to a Topic. The subscription reads the
	// topic's shared log through its own cursor and is registered with the
	// worker as a job source, like a bound queue. Every payload published
	// after this call becomes a job for this worker, processed under its
	// normal strategy and concurrency.
	//
	// A worker subscribes to a topic at most once: subscribing it again while
	// the first subscription is attached panics, since it would process every
	// payload twice.
	//
	// Parameters:
	//   - t Topic[T]: The topic to subscribe to, created with NewTopic.
	//   - configs: Optional, for example WithTopicPriority.
	//
	// Returns:
	//   - Subscription[T]: The subscription handle (Len, Purge, Unsubscribe,
	//     Worker). It has no Add; use Topic.Publish.
	//
	// Example usage:
	//   topic := varmq.NewTopic[Order]()
	//   sub := worker.WithTopic(topic)
	//   topic.Publish(order) // the worker receives a copy
	//   sub.Unsubscribe()    // stop receiving
	WithTopic(t Topic[T], configs ...TopicConfigFunc) Subscription[T]
}

IWorkerBinder is the base interface for binding standard workers to different queue types. It provides methods to connect workers with various queue implementations, enabling flexible worker-queue configurations. This interface extends beyond basic queue operations by supporting both persistence and distribution capabilities, allowing jobs to be durably stored and distributed across multiple instances. With queue binding capabilities for handling tasks of type T without returning results.

func NewWorker

func NewWorker[T any](wf func(j Job[T]), config ...any) IWorkerBinder[T]

NewWorker creates a worker that can be bound to standard, priority, and persistent and distributed queue types. If concurrency in config is not provided, it defaults to 1. and if concurrency is less than 1, it defaults to number of CPU cores. Parameters:

• wf: Worker function that processes queue items
• config: Optional configuration parameters (concurrency, idle worker ratio, etc.)

Example:

worker := NewWorker(func(data string) {
    return len(data), nil
}, 4) // 4 concurrent workers
queue := worker.BindQueue() // Bind to standard queue
priorityQueue := worker.BindPriorityQueue() // Bind to priority queue
persistentQueue := worker.BindPersistentQueue(myPersistentQueue) // Bind to persistent queue
distQueue := worker.BindDistributedQueue(myDistributedQueue) // Bind to distributed queue

type Identifiable

type Identifiable interface {
	// ID returns the unique identifier of the job.
	// Returns empty in case don't set any id by client
	ID() string
}

type Item

type Item[T any] struct {
	ID       string
	Data     T
	Priority int
}

Item represents a data item to be processed by a worker It combines a unique identifier with a value of any generic type and a optional priority for priority queues This only used for AddAll operations

type Job

type Job[T any] interface {
	Identifiable
	// Data returns the data associated with the job.
	Data() T
}

type JobConfigFunc

type JobConfigFunc func(*jobConfigs)

func WithJobId

func WithJobId(id string) JobConfigFunc

WithJobId sets the job ID for the job.

type Metrics

type Metrics interface {
	// Submitted returns the total number of tasks submitted to the worker.
	Submitted() uint64
	// Completed returns the total number of tasks that have completed processing
	// (both successful and failed).
	Completed() uint64
	// Successful returns the total number of tasks that completed successfully.
	Successful() uint64
	// Failed returns the total number of tasks that failed (returned an error or panicked).
	Failed() uint64
	// Reset resets all counters to zero.
	Reset()
	// contains filtered or unexported methods
}

type PendingTracker

type PendingTracker interface {
	// NumPending returns the number of pending jobs in the group.
	NumPending() int
}

type PersistentPriorityQueue

type PersistentPriorityQueue[T any] interface {
	IExternalBaseQueue

	// Worker returns the bound worker.
	Worker() Worker
	// Add adds a new Job with the given priority to the queue
	// It returns true if the job was added successfully
	Add(data T, priority int, configs ...JobConfigFunc) bool
}

type PersistentQueue

type PersistentQueue[T any] interface {
	IExternalBaseQueue

	// Worker returns the bound worker.
	Worker() Worker

	// Add adds a job with the given data to the persistent queue
	// It returns true if the job was added successfully
	Add(data T, configs ...JobConfigFunc) bool
}

PersistentQueue is an interface that extends Queue to support persistent job operations

type PriorityQueue

type PriorityQueue[T any] interface {
	IExternalBaseQueue
	// Worker returns the bound worker.
	Worker() Worker
	// Add adds a new Job with the given priority to the queue and returns a EnqueuedJob to handle the job.
	// Time complexity: O(log n)
	Add(data T, priority int, configs ...JobConfigFunc) (EnqueuedJob, bool)
	// AddAll adds multiple Jobs with the given priority to the queue and returns a EnqueuedGroupJob to handle the job.
	// Time complexity: O(n log n) where n is the number of Jobs added
	AddAll(data []Item[T]) EnqueuedGroupJob
}

type Queue

type Queue[T any] interface {
	IExternalBaseQueue

	// Worker returns the bound worker.
	Worker() Worker
	// Add adds a new Job to the queue and returns a EnqueuedJob to handle the job.
	// Time complexity: O(1)
	Add(data T, configs ...JobConfigFunc) (EnqueuedJob, bool)
	// AddAll adds multiple Jobs to the queue and returns a EnqueuedGroupJob to handle the job.
	// Time complexity: O(n) where n is the number of Jobs added
	AddAll(data []Item[T]) EnqueuedGroupJob
}

type QueueConfigFunc

type QueueConfigFunc func(*queueConfig)

func WithQueueCapacity

func WithQueueCapacity(capacity int) QueueConfigFunc

WithQueueCapacity configures the maximum number of pending items a queue can hold.

When the queue reaches its capacity, subsequent Add calls will return false, indicating the item was not enqueued. This allows callers to implement backpressure or overflow handling. Capacity bounds the jobs in storage: jobs a worker has taken in a batch, or parked on Stop or Pause, no longer count, so up to 32 more per worker can be pending.

Parameters:

• capacity: The maximum number of items the queue can hold. Values less than 1 are treated as 0, meaning unlimited capacity.

Default behavior: If this option is not set, the queue has unlimited capacity (0). Note: If the queue doesn't implement the CapacitySetter interface, this option will be ignored.

func WithQueuePriority

func WithQueuePriority(priority int) QueueConfigFunc

WithQueuePriority configures the priority level of a queue when binding it to a worker.

The worker uses this priority to determine which queue to process jobs from first. Queues with a higher priority (lower integer values) will be processed before queues with a lower priority. In the case of ties, insertion order is preserved.

Parameters:

• priority: An integer representing the priority level. A lower number indicates a higher priority. For example, priority 1 is processed before priority 5.

Examples:

• worker.BindQueue(varmq.WithQueuePriority(1)): Assigns a high priority to the queue.
• worker.BindQueue(varmq.WithQueuePriority(10)): Assigns a lower priority to the queue.

Default behavior: If this option is not set, the queue implicitly receives the lowest possible priority (math.MaxInt).

type Result

type Result[T any] struct {
	JobId string
	Data  T
	Err   error
}

Result represents the result of a job, containing the data and any error that occurred.

type ResultPriorityQueue

type ResultPriorityQueue[T, R any] interface {
	IExternalBaseQueue
	// Worker returns the bound worker.
	Worker() Worker
	// Add adds a new Job with the given priority to the queue and returns a EnqueuedResultJob to handle the job with result receiving.
	// Time complexity: O(log n)
	Add(data T, priority int, configs ...JobConfigFunc) (EnqueuedResultJob[R], bool)
	// AddAll adds multiple Jobs with the given priority to the queue and returns a EnqueuedResultGroupJob to handle the job with result receiving.
	// Time complexity: O(n log n) where n is the number of Jobs added
	AddAll(data []Item[T]) EnqueuedResultGroupJob[R]
}

type ResultQueue

type ResultQueue[T, R any] interface {
	IExternalBaseQueue
	// Worker returns the bound worker.
	Worker() Worker
	// Add adds a new Job to the queue and returns a EnqueuedResultJob to handle the job with result receiving.
	// Time complexity: O(1)
	Add(data T, configs ...JobConfigFunc) (EnqueuedResultJob[R], bool)
	// AddAll adds multiple Jobs to the queue and returns a EnqueuedResultGroupJob to handle the job with result receiving.
	// Time complexity: O(n) where n is the number of Jobs added
	AddAll(data []Item[T]) EnqueuedResultGroupJob[R]
}

type StatusProvider

type StatusProvider interface {
	// IsClosed returns true if the job has been closed.
	IsClosed() bool
	// Status returns a string representation of the job's current lifecycle state.
	// Possible return values:
	// - "Created": Job has been created but not yet added to a queue
	// - "Queued": Job has been added to a queue and is waiting to be processed
	// - "Processing": Job is currently being processed by a worker
	// - "Finished": Job processing has completed successfully
	// - "Closed": Job has been manually closed and will not be processed
	Status() string
}

type Strategy

type Strategy uint8

const (
	// Selects queues in a round-robin fashion
	RoundRobin Strategy = iota
	// Selects the queue with the most items
	MaxLen
	// Selects the queue with the fewest items
	MinLen
	// Selects the queue with the highest priority first
	Priority
)

type Subscription

type Subscription[T any] interface {
	IExternalBaseQueue
	// Worker returns the bound worker.
	Worker() Worker
	// Unsubscribe ends the subscription and detaches it from the worker and
	// drops what this subscription has not read. It is idempotent.
	Unsubscribe()
}

Subscription is a worker's handle on a topic, returned by IWorkerBinder.WithTopic. It has no Add: the topic is the only producer.

Len and Purge act on what this subscription has not read yet. Purge never calls Close on payloads.

IsFull reports whether this subscription has as many unread payloads as the topic's WithTopicCapacity, so it is the one making Publish return false. It is always false without a capacity.

Close detaches this handle. It is the same as Unsubscribe.

type Topic

type Topic[T any] interface {
	// Publish appends data to the topic and wakes every subscriber's worker.
	//
	// It returns false, and no subscriber gets the payload, when there are
	// zero subscriptions, the topic is closed, or the topic is full: its
	// slowest subscription already has WithTopicCapacity unread payloads. To
	// wait for a subscriber to finish its payloads, use its Worker's Wait.
	Publish(data T) bool
	// IsFull reports whether the topic has reached its WithTopicCapacity: its
	// slowest subscription already has that many unread payloads, so Publish
	// would return false. It is always false without a capacity. The answer
	// can change right after the call; Publish's result is authoritative.
	IsFull() bool
	// NumSubscribers returns the number of subscriptions attached in this
	// process.
	NumSubscribers() int
	// Close unsubscribes every subscription and makes further Publish calls
	// return false. It is idempotent and always returns nil.
	Close() error
	// contains filtered or unexported methods
}

Topic fans out every published payload to all current subscriptions.

Semantics are those of a fanout exchange, not a replayable log: a subscriber receives only payloads published after it subscribed.

Payloads are stored once in a shared in-memory log built from linked chunks. Each subscription reads at its own position, so a slow subscriber never blocks the publisher or other subscribers. A chunk is freed once the slowest subscriber has moved past it, so a stalled or paused subscriber holds every chunk published since it stalled until it catches up or unsubscribes.

Each subscriber's worker gets its own job wrapping the payload. The payload is copied by value; if T is a pointer, or contains pointers, maps or slices, subscribers share the pointee. Job IDs are generated when a subscriber's worker reads the payload, using that worker's ID generator.

Create a topic with NewTopic; subscribe a worker with IWorkerBinder.WithTopic.

Example usage:

topic := varmq.NewTopic[Order]()
varmq.NewWorker(sendEmail).WithTopic(topic)
varmq.NewWorker(updateStock, 4).WithTopic(topic)

topic.Publish(order) // both workers get their own job from one shared log

func NewTopic

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

NewTopic creates an in-memory topic. See Topic for its semantics. configs takes topic options such as WithTopicCapacity.

type TopicConfigFunc

type TopicConfigFunc func(*topicConfig)

TopicConfigFunc configures a topic or a worker's subscription to one, like QueueConfigFunc does for queues: NewTopic reads the capacity, WithTopic reads the priority.

func WithTopicCapacity

func WithTopicCapacity(capacity int) TopicConfigFunc

WithTopicCapacity bounds an in-memory topic, passed to NewTopic.

A payload stays in the topic until every subscription has read it, so one slow, paused or starved subscriber makes the topic keep everything published since it fell behind. With a capacity, Publish returns false once the slowest subscription has that many unread payloads, the way Add does on a full queue. The refused payload reaches no subscriber. Room comes back as that subscription reads, purges or unsubscribes.

Parameters:

• capacity: The maximum number of unread payloads for the slowest subscription. Values less than 1 mean unlimited.

Example:

• varmq.NewTopic[Order](varmq.WithTopicCapacity(10_000))

Default behavior: unlimited.

func WithTopicPriority

func WithTopicPriority(priority int) TopicConfigFunc

WithTopicPriority sets the priority of a topic subscription among the worker's sources, the same way WithQueuePriority does for a bound queue.

A worker can read from several topics and queues. Under the default Priority strategy it takes the next job from the source with the lowest number that has pending payloads; ties keep subscription order. Other strategies (RoundRobin, MaxLen, MinLen) ignore it.

Example:

• worker.WithTopic(orders, varmq.WithTopicPriority(1)): read orders first.

Default behavior: without this option the subscription gets the lowest possible priority (math.MaxInt).

type Worker

type Worker interface {
	// Name returns the unique name of the worker.
	Name() string
	// IsRunning returns whether the worker is actively processing jobs.
	//
	// A worker is Running when it has at least one job currently being processed.
	// This is a transient state — when no jobs are processing, the worker
	// transitions to Idle (if active) or Paused/Stopped.
	IsRunning() bool
	// IsIdle returns whether the worker is in the Idle state.
	//
	// A worker is Idle when it is active (started and not paused/stopped)
	// but has no pending jobs in the queue and no jobs currently being processed.
	// It can immediately resume processing when new jobs arrive.
	IsIdle() bool
	// IsActive returns whether the worker is either Running or Idle.
	//
	// An active worker is started and ready to process jobs. It is not
	// Paused, Stopping, Stopped, or in an intermediate transition state.
	IsActive() bool
	// IsPaused returns whether the worker is in the Paused state.
	//
	// A paused worker does not process new jobs from the queue, but any
	// in-flight jobs that were already running continue until completion.
	// Use Resume() to resume processing.
	IsPaused() bool
	// IsStopped returns whether the worker is in the Stopped state.
	//
	// A stopped worker has ceased all processing and removed all worker
	// goroutines. It cannot be resumed; use Restart() to start again.
	IsStopped() bool
	// Status returns the human-readable current status of the worker.
	//
	// Possible values: "Initiated", "Idle", "Running", "Pausing", "Paused",
	// "Stopping", "Stopped"
	Status() string
	// NumProcessing returns the number of jobs being executed right now,
	// never more than NumConcurrency.
	NumProcessing() int
	// NumPending returns the number of jobs that have not started: those
	// still in the bound queues plus those already handed to a worker in a
	// batch and waiting their turn, or parked by a Stop or Pause. A bound
	// queue's own Len counts only the former. A job closed while it waits in
	// storage, a batch or the park counts until the worker reaches and
	// discards it. Submitted == NumPending + NumProcessing + Completed, except
	// for an instant while a stopping worker parks its batch.
	NumPending() int
	// NumConcurrency returns the upper bound on jobs processed at once. It
	// is a limit, not a pool size: goroutines are spawned on demand and only
	// while another one would shorten the backlog by more than it costs.
	NumConcurrency() int
	// NumIdleWorkers returns the number of idle worker goroutines
	// currently available in the pool to process new jobs.
	NumIdleWorkers() int
	// Errs returns a read-only channel that receives all errors
	// encountered during job processing.
	Errs() <-chan error
	// Context returns the context associated with this worker.
	//
	// The context is cancelled when the worker is Stopped or Restarted.
	Context() context.Context
	// Metrics returns the metrics collected for this worker,
	// including submitted, completed, successful, and failed job counts.
	Metrics() Metrics
	// TunePool adjusts the concurrency (pool size) of the worker
	// up or down. This takes effect immediately for new jobs.
	//
	// Errors:
	//   - ErrSameConcurrency: worker already has this concurrency value.
	//   - see getStatusError() for additional state-specific errors.
	TunePool(concurrency int) error
	// Pause initiates a graceful pause of the worker.
	//
	// This is non-blocking and returns immediately. The behavior depends on
	// whether there are in-flight jobs:
	//   - If no jobs are processing: transitions directly to Paused.
	//   - If jobs are processing: transitions to Pausing. In-flight jobs
	//     continue until completion, then the status becomes Paused. Jobs a
	//     worker holds in its batch but has not started are parked on the
	//     worker: still Queued and counted by NumPending, and handed out
	//     first, in order, on Resume.
	//
	// Once Paused, no new jobs are processed from the queue. Use Resume()
	// to resume processing, or PauseAndWait() to block until fully paused.
	//
	// Pause is idempotent: calling Pause on an already Paused or Pausing
	// worker returns nil.
	//
	// Errors:
	//   - ErrWorkerStopped: worker is in the Stopped state.
	//   - ErrWorkerStopping: worker is in the Stopping state.
	//   - ErrNotRunningWorker: worker is in the Initiated state.
	Pause() error

	// Stop initiates a graceful shutdown of the worker.
	//
	// This is non-blocking and returns immediately. The worker transitions
	// to Stopping and cancels its internal context. The event loop then
	// waits for in-flight jobs to finish, removes all worker goroutines,
	// and transitions to Stopped. Jobs a worker holds in its batch but has
	// not started are parked on the worker: still Queued and counted by
	// NumPending, and handed out first, in order, on Restart.
	//
	// A stopped worker cannot be resumed; use Restart() instead.
	// Use StopAndWait() to block until fully stopped.
	//
	// Stop is idempotent: calling Stop on an already Stopped or Stopping
	// worker returns nil.
	//
	// Errors:
	//   - ErrNotRunningWorker: worker is in the Initiated state.
	Stop() error
	// Restart gracefully restarts the worker, preserving any pending
	// jobs in the queue. It stops the worker, waits for it to reach
	// Stopped, recreates the internal context, and starts again.
	//
	// Restart works from any valid state: Running, Idle, Pausing, Paused,
	// Stopping, or Stopped. It transparently handles the case where another
	// goroutine already restarted the worker by returning ErrRunningWorker.
	//
	// Errors:
	//   - ErrRunningWorker: worker is already active (concurrent restart).
	//   - ErrNotRunningWorker: worker is in the Initiated state.
	//   - Errors from Stop(): see Stop's error documentation.
	Restart() error
	// Resume resumes a paused worker and begins processing pending
	// jobs from the queue.
	//
	// Resume is idempotent: calling Resume on an already active (Running or
	// Idle) worker returns nil.
	//
	// Errors:
	//   - ErrWorkerStopped: worker is in the Stopped state. Use Restart().
	//   - ErrWorkerStopping: worker is in the Stopping state.
	//   - ErrWorkerPausing: worker is in the Pausing state.
	//   - ErrNotRunningWorker: worker is in the Initiated state.
	Resume() error
	// Start begins processing jobs from the bound queues.
	// When WithAutoRun is true (the default), the worker starts automatically
	// after creation and this method returns ErrRunningWorker.
	// Use WithAutoRun(false) to create a worker in an initiated state,
	// then call Start() manually when ready.
	//
	// Start only accepts initiated or stopped states. For all other states
	// that still have a running event loop, it returns the corresponding
	// status error.
	//
	// Errors:
	//   - ErrRunningWorker: worker is already active (Running or Idle).
	//   - ErrWorkerPaused: worker is in the Paused state. Use Resume().
	//   - ErrWorkerStopping: worker is in the Stopping state.
	//   - ErrWorkerPausing: worker is in the Pausing state.
	Start() error
	// Wait blocks until the worker has no pending jobs in the queue and no jobs
	// currently being processed.
	//
	// This returns when:
	//   - Status is Running or Idle: queue is empty AND curProcessing == 0
	//   - Status is Paused or Stopped: curProcessing == 0 (all in-flight jobs done)
	//   - Status is Pausing or Stopping: curProcessing == 0
	//
	// Unlike WaitUntilIdle(), this does NOT require the status to be Idle.
	// For example, if Pause() is called and all jobs drain, Wait() returns
	// even though the status becomes Paused (not Idle).
	//
	// Use this when you only care that work is done, regardless of final status.
	// Use WaitUntilIdle() when you specifically need the Idle state.
	Wait()
	// WaitUntilIdle blocks until the worker reaches Idle status with no pending jobs.
	//
	// Unlike Wait(), which returns as soon as all in-flight jobs complete and the
	// queue is drained (regardless of final status), WaitUntilIdle specifically
	// waits for the worker's status to become Idle with an empty queue.
	//
	// This means:
	//   - If status is Running with no work, it blocks until the worker transitions to Idle
	//   - If status is Paused with no work, it blocks indefinitely (Paused != Idle)
	//   - If status is Stopped, it blocks indefinitely (Stopped != Idle)
	//
	// Use this when you need the worker in the Idle state specifically.
	// Use Wait() when you only care that work is done.
	//
	// Note: If the worker is already Idle with no pending jobs, this returns immediately.
	WaitUntilIdle()
	// WaitUntilPaused blocks until the worker reaches Paused status specifically.
	//
	// This blocks indefinitely if the worker never transitions to Paused.
	// For example, if Restart() or Stop() is called instead of Pause(),
	// the status will never become Paused and this will hang.
	//
	// Note: If the worker is already Paused, this returns immediately.
	WaitUntilPaused()
	// WaitUntilStopped blocks until the worker reaches Stopped status specifically.
	//
	// This blocks indefinitely if the worker never transitions to Stopped.
	// For example, if Pause() is called instead of Stop(), the status will
	// never become Stopped and this will hang.
	//
	// Note: If the worker is already Stopped, this returns immediately.
	WaitUntilStopped()
	// WaitAndStop waits until all pending jobs are processed and in-flight
	// jobs complete, then initiates a graceful stop and returns immediately.
	//
	// This is equivalent to calling Wait() followed by Stop().
	// Use this when you want to drain the queue before shutting down.
	// Use StopAndWait() when you need to block until the worker reaches
	// the Stopped state.
	//
	// Errors:
	//   - see Stop() for error documentation.
	WaitAndStop() error
	// StopAndWait initiates a graceful stop and blocks until the worker
	// reaches the Stopped state.
	//
	// This is equivalent to calling Stop() followed by WaitUntilStopped().
	// If there are in-flight jobs, this blocks until they complete and
	// all worker goroutines are removed.
	//
	// Errors:
	//   - see Stop() for error documentation.
	StopAndWait() error
	// WaitAndPause waits until all pending jobs are processed and in-flight
	// jobs complete, then pauses the worker.
	//
	// This is equivalent to calling Wait() followed by Pause(). Use this
	// when you want to drain the queue before pausing.
	//
	// Errors:
	//   - see Pause() for error documentation.
	WaitAndPause() error
	// PauseAndWait initiates a pause and blocks until the worker reaches
	// the Paused state.
	//
	// This is equivalent to calling Pause() followed by WaitUntilPaused().
	// If there are in-flight jobs, this blocks until they complete and
	// the status transitions from Pausing to Paused.
	//
	// Errors:
	//   - see Pause() for error documentation.
	PauseAndWait() error
	// contains filtered or unexported methods
}

Worker represents a worker that processes Jobs.