Help us shape the future of Go! Take the 2026 Go Developer Survey

varmq

package module
v1.3.0 Latest Latest
Warning

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

 Go to latest Published: Jun 21, 2025 License: MIT Imports: 12 Imported by: 0

README

VarMQ

Go Reference DeepWiki Playground Go Report Card CI Codecov 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
  • 🧩 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
  defer worker.WaitUntilFinished()
  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 three strategies:

  1. RoundRobin (default - cycles through queues equally)
  2. MaxLen (prioritizes queues with more jobs)
  3. 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
defer worker.WaitUntilFinished()

// Bind to a standard queues
q1 := worker.BindQueue()
q2 := worker.BindQueue()
pq := worker.BindPriorityQueue()

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

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
}

↗️ Run it on Playground

It will process jobs from all queues in a round-robin fashion.

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.WaitUntilFinished()
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 only returns 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.WaitUntilFinished()

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.

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 918.6 128 3
Priority 952.7 144 4
ErrWorker ErrQueue 1017 305 6
ErrPriority 1006 320 7
ResultWorker ResultQueue 1026 353 6
ResultPriority 1039 368 7
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 635,186 146,841 4,002
Priority 755,276 162,144 5,002
ErrWorker ErrQueue 673,912 171,090 4,505
ErrPriority 766,043 186,663 5,505
ResultWorker ResultQueue 675,420 187,897 4,005
ResultPriority 777,680 203,263 5,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 635ns.

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
Charts
Metric Add Operation AddAll Operation
Execution Time
Time (ns/op) VarMQ Add/Execute Benchmark Chart
Time (ms/op) VarMQ AddAll/Execute Benchmark Chart
Memory Usage
Memory (B/op) VarMQ Add/Memory Benchmark Chart
Memory (KB/op) VarMQ AddAll/Memory Benchmark Chart
Allocations
Allocations (allocs/op) VarMQ Add/Allocations Benchmark Chart
Allocations (allocs/op) VarMQ AddAll/Allocations Benchmark Chart

Chart images is been generated using Vizb

Comparison with other packages

API Reference

For detailed API documentation, see the API Reference.

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

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

This section is empty.

Functions

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 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 ConfigFunc 

type ConfigFunc func(*configs)

ConfigFunc is a function that configures a worker.

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

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

Parameters:

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

Default: If this option is not set, RoundRobin 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) 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) 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.
	// 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() 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.
	//
	// 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) 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() 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.
	//
	// 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) 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 {
	PendingTracker
	// 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.
	//
	// 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() 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.
	//
	// 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) 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.
	//
	// 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() 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.
	//
	// 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) 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.
	//
	// 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() 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.
	//
	// 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) 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.
	//
	// 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() 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.
	//
	// 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) 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 IQueue: A queue implementation that provides persistence capabilities.
	//
	// 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) 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.
	//
	// 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) PersistentPriorityQueue[T]

	// WithDistributedQueue binds the void 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, RabbitMQ, Kafka, or any other distributed messaging system.
	//
	// 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) DistributedQueue[T]

	// WithDistributedPriorityQueue binds the void 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.
	//
	// 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) 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 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 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
)

type Worker 

type Worker interface {
	// IsPaused returns whether the worker is paused.
	IsPaused() bool
	// IsStopped returns whether the worker is stopped.
	IsStopped() bool
	// IsRunning returns whether the worker is running.
	IsRunning() bool
	// Status returns the current status of the worker.
	Status() string
	// NumProcessing returns the number of Jobs currently being processed by the worker.
	NumProcessing() int
	// NumConcurrency returns the current concurrency or pool size of the worker.
	NumConcurrency() int
	// NumIdleWorkers returns the number of idle workers in the pool.
	NumIdleWorkers() int
	// TunePool tunes (increase or decrease) the pool size of the worker.
	TunePool(concurrency int) error
	// Pause pauses the worker.
	Pause() error
	// PauseAndWait pauses the worker and waits until all ongoing processes are done.
	PauseAndWait() error
	// Stop stops the worker and waits until all ongoing processes are done to gracefully close the channels.
	Stop() error
	// Restart restarts the worker and initializes new worker goroutines based on the concurrency.
	Restart() error
	// Resume continues processing jobs those are pending in the queue.
	Resume() error
	// WaitUntilFinished waits until all pending Jobs in the are processed.
	WaitUntilFinished()
	// WaitAndStop waits until all pending Jobs in the queue are processed and then closes the queue.
	WaitAndStop() error
	// contains filtered or unexported methods
}

Worker represents a worker that processes Jobs.

Source Files

View all Source files

Directories

Path Synopsis
examples
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.