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

varmq

package module
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: May 1, 2025 License: MIT Imports: 9 Imported by: 0

README

VarMQ

Go Reference Go Report Card Go Version CI codecov CodeRabbit Pull Request Reviews License

VarMQ is a high-performance message queue for Go that simplifies concurrent task processing. It combines a message queue with worker pool management. 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:
    • WorkerFunc - Returns result and error
    • WorkerErrFunc - Returns only error (when result isn't needed)
    • VoidWorkerFunc - Fire-and-forget operations (most performant)
  • 🚦 Concurrency control: Fine-grained control over worker pool size
  • 💾 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() {
    // Create a worker that processes strings and returns their length
    worker := varmq.NewWorker(func(data string) (int, error) {
        fmt.Println("Processing:", data)
        time.Sleep(1 * time.Second) // Simulate work
        return len(data), nil
    })

    // Bind to a standard queue
    queue := worker.BindQueue()
    defer queue.WaitAndClose() // Wait for all jobs to complete and close the queue

    // Add jobs to the queue
    job1 := queue.Add("Hello")
    job2 := queue.Add("World")

    // Get results (Result() returns both value and error)
    result1, err1 := job1.Result()
    if err1 != nil {
        fmt.Println("Error processing job1:", err1)
    } else {
        fmt.Println("Result 1:", result1)
    }

    result2, err2 := job2.Result()
    if err2 != nil {
        fmt.Println("Error processing job2:", err2)
    } else {
        fmt.Println("Result 2:", result2)
    }

    // Add multiple jobs at once
    groupJob := queue.AddAll([]varmq.Item[string]{
        {Value: "Concurrent", ID: "1"},
        {Value: "Queue", ID: "2"},
    })

    resultChan, err := groupJob.Results()
    if err != nil {
        fmt.Println("Error getting results channel:", err)
        return
    }

    // Get results as they complete in real-time
    for result := range resultChan {
        if result.Err != nil {
            fmt.Printf("Error processing job %s: %v\n", result.JobId, result.Err)
        } else {
            fmt.Printf("Result for job %s: %v\n", result.JobId, result.Data)
        }
    }
}

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)
Job Control
// Add a job with custom ID
job := queue.Add("Important task", varmq.WithJobId("custom-id-123"))

// Get job status
status := job.Status()

// Get job result (blocks until job completes)
result, err := job.Result()
if err != nil {
    fmt.Println("Job failed:", err)
} else {
    fmt.Println("Result:", result)
}

// Drain the result (when you don't need it)
job.Drain()

// Get JSON representation of job
jsonData, _ := job.Json()

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 implementingVarMQ's internal queue interfaces
    • Currently supports Redis using redisq adapter
    • Future plans include SQLite, PostgreSQL, DiceDB and more
    • Build your own adapters for any persistent storage system

API Reference

For detailed API documentation, see the API Reference.

Table of Contents

The Concurrency Architecture

varmq architecture

Sequence Diagram

The following sequence diagram illustrates the main flow and interactions in the VarMQ.

sequenceDiagram
    participant Client
    participant Worker as Worker
    participant Queue as Queue
    participant Job as Job
    participant ResultChannel as ResultChannel

    %% Worker Creation
    Client->>Worker: NewWorker(WorkerFunc, config)
    activate Worker
    Note over Worker: Create worker with concurrency settings, channels stack

    %% Queue Binding
    Client->>Worker: BindQueue()
    Worker->>Worker: start()
    activate Worker
    Worker->>Worker: Create channels stack
    Worker->>Worker: spawnWorker for each channel
    Worker->>Worker: startEventLoop()
    Note over Worker: Event loop continuously checks for pending jobs and available worker capacity
    deactivate Worker
    Worker->>Queue: newQueue(worker, internalQueue)
    Queue-->>Worker: Return queue instance
    Queue-->>Client: Return queue

    %% Adding a Job
    Client->>Queue: Add(data, configs)
    activate Queue
    Queue->>Job: newJob(data, configs)
    activate Job
    Job-->>Queue: Return job
    Queue->>Queue: internalQueue.Enqueue(job)
    Queue->>Job: ChangeStatus(queued)
    Queue->>Worker: notifyToPullNextJobs()
    Note over Worker: Signal event loop to check for jobs
    deactivate Queue

    %% Job Processing
    Worker->>Worker: processNextJob()
    activate Worker
    Worker->>Queue: Queue.Dequeue()
    Queue-->>Worker: Return job
    Worker->>Job: ChangeStatus(processing)
    Worker->>Worker: pickNextChannel() <- job
    deactivate Worker

    Worker->>Worker: spawnWorker(channel)
    activate Worker
    Worker->>Worker: Execute worker function with job.Input
    alt Success
        Job->>ResultChannel: Send(result)
    else Error
        Worker->>Job: SaveAndSendError(err)
        Job->>ResultChannel: Send(error)
    end
    Worker->>Worker: CurProcessing.Add(-1)
    Worker->>Worker: notifyToPullNextJobs()
    deactivate Worker

    %% Job Completion
    Worker->>Job: ChangeStatus(finished)
    Worker->>Job: close()
    Job->>ResultChannel: close()
    deactivate Job

    %% Result Handling
    Client->>Job: Result()
    activate Job
    Job->>ResultChannel: <-ch
    ResultChannel-->>Job: Return result
    Job-->>Client: Return result, error
    deactivate Job

    %% Worker Control
    Client->>Worker: Pause()
    Client->>Worker: Resume()
    Client->>Worker: Stop()
    deactivate Worker

    %% Queue Shutdown
    Client->>Queue: WaitAndClose()
    activate Queue
    Queue->>Queue: Wait for all jobs to complete
    Queue->>Worker: Wait for worker to finish
    Queue->>Queue: Close()
    Queue->>Worker: Stop()
    Queue-->>Client: Return when complete
    deactivate Queue

    alt Direct Close
        Client->>Queue: Close()
        activate Queue
        Queue->>Queue: Discard pending jobs
        Queue->>Worker: Stop()
        Queue-->>Client: Return
        deactivate Queue
    end

    %% Queue Purge
    Client->>Queue: Purge()
    activate Queue
    Queue->>Queue: internalQueue.Purge()
    Note over Queue: All pending jobs are removed
    Queue-->>Client: Return
    deactivate Queue

Older Versions

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 ConfigFunc 

type ConfigFunc func(*configs)

func WithAutoCleanupCache 

func WithAutoCleanupCache(duration time.Duration) ConfigFunc

func WithCache 

func WithCache(cache ICache) ConfigFunc

func WithConcurrency 

func WithConcurrency(concurrency int) ConfigFunc

func WithJobIdGenerator 

func WithJobIdGenerator(fn func() string) ConfigFunc

type DistributedPriorityQueue 

type DistributedPriorityQueue[T, R any] interface {
	IExternalBaseQueue
	// Time complexity: O(log n)
	Add(data T, priority int, configs ...JobConfigFunc) bool
}

func NewDistributedPriorityQueue 

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

type DistributedQueue 

type DistributedQueue[T, R any] interface {
	IExternalBaseQueue
	// Time complexity: O(1)
	Add(data T, configs ...JobConfigFunc) bool
}

func NewDistributedQueue 

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

type EnqueuedGroupJob 

type EnqueuedGroupJob[T any] interface {
	// Drain discards the job's result and error values asynchronously.
	Drain() error
	// Results returns a channel that will receive the results of the group
	Results() (<-chan Result[T], error)
}

type EnqueuedJob 

type EnqueuedJob[R any] interface {
	Job
	// Drain discards the job's result and error values asynchronously.
	Drain() error
	// Result blocks until the job completes and returns the result and any error.
	Result() (R, error)
}

EnqueuedJob represents a job that has been enqueued and can wait for a result.

type EnqueuedSingleGroupJob 

type EnqueuedSingleGroupJob[R any] interface {
	Job
	EnqueuedGroupJob[R]
}

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 ICache 

type ICache interface {
	Load(key any) (any, bool)
	Store(key any, value any)
	Delete(key any)
	Range(f func(key, value any) bool)
	Clear()
}

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 IExternalBaseQueue 

type IExternalBaseQueue interface {
	// PendingCount returns the number of Jobs pending in the queue.
	PendingCount() int
	// Purge removes all pending Jobs from the queue.
	// Time complexity: O(n) where n is the number of pending Jobs
	Purge()
	// Close closes the queue and resets all internal states.
	// Time complexity: O(n) where n is the number of channels
	Close() error
}

type IExternalQueue 

type IExternalQueue[T, R any] interface {
	IExternalBaseQueue

	// Worker returns the worker.
	Worker() Worker[T, R]
	// JobById returns the job with the given id.
	JobById(id string) (EnqueuedJob[R], error)
	// GroupsJobById returns the groups job with the given id.
	GroupsJobById(id string) (EnqueuedSingleGroupJob[R], error)
	// WaitUntilFinished waits until all pending Jobs in the queue are processed.
	// Time complexity: O(n) where n is the number of pending Jobs
	WaitUntilFinished()
	// WaitAndClose waits until all pending Jobs in the queue are processed and then closes the queue.
	// Time complexity: O(n) where n is the number of pending Jobs
	WaitAndClose() error
}

IExternalQueue is the root interface of concurrent queue operations.

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 ISubscribable 

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

ISubscribable is the root interface of subscribable operations.

type IVoidWorkerBinder 

type IVoidWorkerBinder[T any] interface {
	IWorkerBinder[T, any]

	// 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:
	//   distributedQueue := NewDistributedQueue() // satisfies IDistributedQueue interface,  might be backed by Redis or other systems
	//   distributedQueue := voidWorker.WithDistributedQueue(distributedQueue)
	//   distributedQueue.Add(data) // This job can be processed by any worker instance listening to this queue
	WithDistributedQueue(dq IDistributedQueue) DistributedQueue[T, any]

	// 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, any]
}

IVoidWorkerBinder extends IWorkerBinder with distributed queue capabilities specifically for void workers that don't return results. This interface is specialized for workers that perform actions but don't need to return any data, making them suitable for distributed processing scenarios where results aren't needed or are handled externally.

The void worker pattern is ideal for fire-and-forget operations like sending notifications, updating external systems, or logging events where no response is required.

func NewVoidWorker 

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

NewVoidWorker creates a worker for operations that don't return any value (void functions). This is the most performant worker type as it doesn't use result channels except for panic handling. VoidWorker is the only worker type that can be bound to distributed queues in addition to standard, priority, and persistent queue types.

Parameters:

  • wf: Worker function that processes items without returning anything
  • config: Optional configuration parameters (concurrency, cache settings, etc.)

If concurrency in config is less than 1 or not provided, it defaults to number of CPU cores.

Example: 

worker := NewVoidWorker(func(data int) {
    fmt.Printf("Processing: %d\n", data)
})
queue := worker.BindQueue() // Bind to standard queue
distQueue := worker.Copy().WithDistributedQueue(myDistributedQueue) // Bind to provided distributed queue

type IWorkerBinder 

type IWorkerBinder[T, R any] interface {
	Worker[T, R]

	// 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, R]: 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, 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:
	//   - Queue[T, R]: 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, 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:
	//   - PriorityQueue[T, R]: 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, 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:
	//   - PriorityQueue[T, R]: 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, R]

	// 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, R]: 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, R]

	// 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, R]: 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, R]
}

IWorkerBinder 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. This interface extends the Worker interface with queue binding capabilities for handling tasks of type T and producing results of type R.

func NewErrWorker 

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

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. Like NewWorker, it can be bound to standard, priority, and persistent queue types.

Parameters:

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

If concurrency in config is less than 1 or not provided, it defaults to number of CPU cores.

Example: 

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

func NewWorker 

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

NewWorker creates a worker that can be bound to standard, priority, and persistent queue types. It accepts a worker function that processes items of type T and returns results of type R.

Parameters:

  • wf: Worker function that processes queue items and returns a result and error
  • config: Optional configuration parameters (concurrency, cache settings, etc.)

If concurrency in config is less than 1 or not provided, it defaults to number of CPU cores.

Example: 

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

type Item 

type Item[T any] struct {
	ID    string
	Value T
}

Item represents a data item to be processed by a worker It combines a unique identifier with a value of any generic type

type Job 

type Job interface {
	// ID returns the unique identifier of the job.
	ID() string
	// IsClosed returns whether the job is closed.
	IsClosed() bool
	// Status returns the current status of the job.
	Status() string
	// Json returns the JSON representation of the job.
	Json() ([]byte, error)
	// contains filtered or unexported methods
}

type JobConfigFunc 

type JobConfigFunc func(*jobConfigs)

func WithJobId 

func WithJobId(id string) JobConfigFunc

type PQItem 

type PQItem[T any] struct {
	// ID is an optional identifier for the item
	ID string
	// Value contains the actual data stored in the queue
	Value T
	// Priority determines the item's position in the queue
	Priority int
}

type PersistentPriorityQueue 

type PersistentPriorityQueue[T, R any] interface {
	PriorityQueue[T, R]
}

type PersistentQueue 

type PersistentQueue[T, R any] interface {
	Queue[T, R]
}

PersistentQueue is an interface that extends Queue to support persistent job operations where jobs can be recovered even after application restarts. All jobs must have unique IDs.

type PriorityQueue 

type PriorityQueue[T, R any] interface {
	IExternalQueue[T, R]
	// Add adds a new Job with the given priority to the queue and returns a channel to receive the result.
	// Time complexity: O(log n)
	Add(data T, priority int, configs ...JobConfigFunc) EnqueuedJob[R]
	// AddAll adds multiple Jobs with the given priority to the queue and returns a channel to receive all responses.
	// Time complexity: O(n log n) where n is the number of Jobs added
	AddAll(items []PQItem[T]) EnqueuedGroupJob[R]
}

type Queue 

type Queue[T, R any] interface {
	IExternalQueue[T, R]
	// 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[R]
	// 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[R]
}

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 VoidWorkerFunc 

type VoidWorkerFunc[T any] func(T)

VoidWorkerFunc represents a function that processes a Job and returns nothing.

type Worker 

type Worker[T, R any] 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
	// Pause pauses the worker.
	Pause() Worker[T, R]
	// Copy returns a copy of the worker.
	Copy(config ...any) IWorkerBinder[T, R]
	// PauseAndWait pauses the worker and waits until all ongoing processes are done.
	PauseAndWait()
	// Stop stops the worker and waits until all ongoing processes are done to gracefully close the channels.
	// Time complexity: O(n) where n is the number of channels
	Stop()
	// Status returns the current status of the worker.
	// Time complexity: O(1)
	Status() string
	// Restart restarts the worker and initializes new worker goroutines based on the concurrency.
	// Time complexity: O(n) where n is the concurrency
	Restart() error
	// Resume continues processing jobs those are pending in the queue.
	// Time complexity: O(n) where n is the concurrency
	Resume() error
	// CurrentProcessingCount returns the number of Jobs currently being processed by the worker.
	// Time complexity: O(1)
	CurrentProcessingCount() uint32
}

Worker represents a worker that processes Jobs.

type WorkerErrFunc 

type WorkerErrFunc[T any] func(T) error

WorkerErrFunc represents a function that processes a Job and returns an error.

type WorkerFunc 

type WorkerFunc[T, R any] func(T) (R, error)

WorkerFunc represents a function that processes a Job and returns a result and an error.

Source Files

View all Source files

Directories

Path Synopsis
examples
server command
void-worker command
internal
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.