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

varmq

package module
v1.2.0 Latest Latest
Warning

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

 Go to latest Published: May 29, 2025 License: MIT Imports: 12 Imported by: 0

README

VarMQ

Go Reference DeepWiki Playground Go Report Card Go Version CI codecov License

VarMQ is a high-performance message queue for Go that simplifies concurrent task processing using worker pool. Using 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 when needed. All through a clean, intuitive API that feels natural to Go developers.

This isn't meant to replace RabbitMQ or Kafka - VarMQ serves a different purpose as a lightweight, in-process message queue with strong worker management. For persistence and distribution, it offers a flexible adapter system that extends its capabilities beyond simple in-memory queues.

✨ Features

  • ⚡ High performance: Optimized for throughput control with minimal overhead
  • 🛠️ 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
  • 💾 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 simple interfaces

Quick Start

Installation
go get github.com/goptics/varmq
Basic Usage
package main

import (
    "fmt"
    "time"

    "github.com/goptics/varmq"
)

func main() {
  w := varmq.NewWorker(func(j varmq.Job[int]) {
    fmt.Printf("Processing %d\n", j.Data())
    time.Sleep(1 * time.Second)
  }, 10)
  defer w.WaitUntilFinished()
  q := w.BindQueue()

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

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.

Advanced Features

Priority Queues

Process important jobs first:

// Create a standard priority queue
queue := worker.BindPriorityQueue()

// Add jobs with priorities (lower number = higher priority)
queue.Add("High priority", 1)
queue.Add("Low priority", 10)

Benchmarks

goos: linux
goarch: amd64
pkg: github.com/goptics/varmq
cpu: AMD EPYC 7763 64-Core Processor
Benchmark Operation Time (ns/op) Memory (B/op) Allocations (allocs/op)
Queue Add 1217 112 3
Queue AddAll (1000 items) 810354 130185 4002
PriorityQueue Add 1296 144 4
PriorityQueue AddAll (1000 items) 1078373 162177 5002
ErrWorker Add 1391 288 6
ErrWorker AddAll (1000 items) 881515 154713 4505
ErrPriorityQueue Add 1452 320 7
ErrPriorityQueue AddAll (1000 items) 1182968 186706 5505
ResultWorker Add 1354 336 6
ResultWorker AddAll (1000 items) 864143 171320 4005
ResultPriorityQueue Add 1450 368 7
ResultPriorityQueue AddAll (1000 items) 1151502 203314 5005

Note: AddAll benchmarks were performed by adding 1000 items in a single call. The reported ns/op, B/op, and allocs/op for AddAll are for the entire batch operation. To estimate per-item metrics for an AddAll operation, divide the table values by 1000 (e.g., for Queue AddAll, the average time per item is approximately 810 ns).

WhyVarMQ?

  • Simple API: Clean, intuitive interface that doesn't get in your way
  • Minimal Dependencies: Core library has no external dependencies
  • Production Ready: Built for real-world scenarios and high-load applications
  • Highly Extensible: Create your own storage adapters by implementing VarMQ's internal queue interfaces
Built-in adapters
  • ⚡ Redis: redisq
  • 🗃️ SQLite: sqliteq
  • 🦆 DuckDB: duckq
  • 🐘 PostgreSQL: 🔄 Upcoming

API Reference

For detailed API documentation, see the API Reference.

The Concurrency Architecture

VarMQ primarily uses its own Event-Loop internally to handle concurrency.

This event loop checks if there are any pending jobs in the queue and if any workers are available in the worker pool. If there are, it distributes jobs to all available workers and then goes back into sleep mode.

When a worker becomes free, it then send pull job request to the event loop.

The event loop then checks again if there are any pending jobs in the queue. If there are, it continues to distribute them to the workers. otherwise, the idle workers been removed from the pool automatically or stay based on WorkerConfig.

varmq architecture

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

👤 Author

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

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.

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 and resets all internal states.
	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
	// 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 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
result-worker command
tune-worker command
internal
helpers
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.