varmq

package module
v1.8.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 4, 2026 License: MIT Imports: 16 Imported by: 0

README

VarMQ

Mentioned in Awesome Go Go Reference Go Report Card CI Codecov Bench Logs Go Version License

A high-performance message queue and pool system for Go that simplifies concurrent task processing using worker pool. Through Go generics, it provides type safety without sacrificing performance.

With VarMQ, you can process messages asynchronously, handle errors properly, store data persistently, and scale across systems using adapters. All through a clean, intuitive API that feels natural to Go developers.

✨ Features

  • ⚡ High performance: Optimized for high throughput with minimal overhead, even under heavy load. see benchmarks
  • 🛠️ Variants of queue types:
    • Standard queues for in-memory processing
    • Priority queues for importance-based ordering
    • Persistent queues for durability across restarts
    • Distributed queues for processing across multiple systems
  • 🧩 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
  • 💾 Persistence: Support for durable storage through adapter interfaces
  • 🌐 Distribution: Scale processing across multiple instances via adapter interfaces
  • 📡 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(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

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

💡 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:

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 a standard queues with coronological priorities
// 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

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 jobs to the queue (non-blocking)
if job, ok := queue.Add("The length of this string is 31"); ok {
 fmt.Println("Job 1 added to queue.")

 go func() {
  result, _ := job.Result()
  fmt.Println("Result:", result)
 }()
}

if job, ok := queue.Add("error"); ok {
 fmt.Println("Job 2 added to queue.")

 go func() {
  _, err := job.Result()
  fmt.Println("Result:", err)
 }()
}

↗️ 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.

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.

Benchmarks

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

Add Operation

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,045 113 2
Priority 1,072 128 3
ErrWorker ErrQueue 1,153 289 5
ErrPriority 1,145 304 6
ResultWorker ResultQueue 1,167 337 5
ResultPriority 1,155 352 6

AddAll Operation

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

Worker Type Queue Type Time (ns/op) Memory (B/op) Allocations (allocs/op)
Worker Queue 649,822 130,601 3,002
Priority 776,409 146,139 4,002
ErrWorker ErrQueue 684,882 154,895 3,505
ErrPriority 808,360 170,661 4,505
ResultWorker ResultQueue 690,249 171,540 3,005
ResultPriority 803,115 187,262 4,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 650ns.

Why is AddAll faster than individual Add calls? Here's what makes the difference:

  1. Batch Processing: Uses a single group job to process multiple items, reducing per-item overhead
  2. Shared Resources: Utilizes a single result channel for all items in the batch

Comparison with Other Packages

We conducted comprehensive benchmarking between VarMQ and Pond v2, as both packages provide similar worker pool functionalities. While VarMQ draws inspiration from some of Pond's design patterns, VarMQ offers unique advantages in queue management and persistence capabilities.

Key Differences:

  • Queue Types: VarMQ provides multiple queue variants (standard, priority, persistent, distributed) vs Pond's single pool type
  • Multi-Queue Management: VarMQ supports binding multiple queues to a single worker with configurable strategies (Priority, RoundRobin, MaxLen, MinLen)

For detailed performance comparisons and benchmarking results, visit:

API Reference

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

The Concurrency Architecture

VarMQ's concurrency model is built around a smart event loop that keeps everything running smoothly.

The event loop continuously monitors for pending jobs in queues and available workers in the pool. When both conditions are met, jobs get distributed to workers instantly. When there's no work to distribute, the system enters a low-power wait state.

Workers operate independently - they process jobs and immediately signal back when they're ready for more work. This triggers the event loop to check for new jobs and distribute them right away.

The system handles worker lifecycle automatically. Idle workers either stay in the pool or get cleaned up based on your configuration, so you never waste resources or run short on capacity.

varmq architecture

Star History

Star History Chart

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

This section is empty.

Variables

View Source 
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")
)
View Source 
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")
)
View Source 
var WorkerRegistry sync.Map

WorkerRegistry safely stores reference to all active workers by Name

Functions

func DefaultConcurrency added in v1.5.0 

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 added in v1.5.0 

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 added in v1.8.0 

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 added in v1.5.0 

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 added in v1.5.0 

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 added in v1.5.0 

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 added in v1.7.0 

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 added in v1.5.0 

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 added in v1.3.0 

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 added in v1.3.0 

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 added in v1.7.0 

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)

func NewServerMux added in v1.7.0 

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 added in v1.3.0 

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 added in v1.2.0 

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 added in v1.6.0 

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 added in v1.7.0 

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 added in v1.4.0 

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 added in v1.8.0 

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 added in v1.2.0 

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 added in v1.2.0 

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 added in v1.7.0 

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 added in v1.7.0 

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 added in v1.3.0 

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 added in v1.2.0 

type Drainer interface {
	// Drain cleans up the job's internal communication channels and releases associated resources.
	// This should be called when the job is no longer needed.
	// After calling Drain, the job's result channels will be closed and cannot be used again.
	Drain()
}

type EnqueuedErrGroupJob added in v1.2.0 

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 added in v1.2.0 

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 added in v1.2.0 

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 added in v1.2.0 

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 added in v1.2.0 

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 added in v1.2.0 

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 added in v1.2.0 

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.
	Len() int
	// IsFull returns true if the queue has reached its configured capacity.
	// If no capacity is set, it always returns false.
	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 added in v1.2.0 

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 added in v1.2.0 

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]
}

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 added in v1.2.0 

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 added in v1.4.0 

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 added in v1.2.0 

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 added in v1.5.0 

type QueueConfigFunc func(*queueConfig)

func WithQueueCapacity added in v1.6.0 

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.

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 added in v1.5.0 

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 added in v1.2.0 

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 added in v1.2.0 

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 added in v1.2.0 

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 added in v1.3.0 

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 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 currently being processed
	// by the worker (in-flight jobs).
	NumProcessing() int
	// NumPending returns the number of jobs waiting in the queue
	// that have not yet started processing.
	NumPending() int
	// NumConcurrency returns the maximum number of jobs that can be
	// processed concurrently (the pool size).
	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.
	//
	// 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.
	//
	// 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.

Source Files

View all Source files

Directories

Path Synopsis
examples
api-control command
idle-wokers command
multi-queue-binds command
result-worker command
tune-worker command
internal
helpers
linkedbuffer
linkedlist
pool
queues

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.