queue

func NewEmptyLogger ¶

func NewEmptyLogger() Logger

NewEmptyLogger creates a no-op logger that discards all log messages. This is useful for:

• Performance-sensitive production environments where logging overhead matters
• Testing scenarios where log output would clutter test results
• Silent background workers that don't need observability

Example:

q := queue.NewPool(5, queue.WithLogger(queue.NewEmptyLogger()))

l := NewEmptyLogger()
l.Info("test")
l.Infof("test")
l.Error("test")
l.Errorf("test")
l.Fatal("test")
l.Fatalf("test")

func NewLogger ¶

func NewLogger() Logger

NewLogger creates a standard logger that writes to stderr with timestamps. This is the default logger used by queues unless overridden with WithLogger.

Log format:

• INFO messages: Simple timestamped output
• ERROR messages: Simple timestamped output
• FATAL messages: Includes stack trace with file:line information

Use cases:

• Development and debugging
• Simple production deployments without structured logging
• When detailed error context is needed

func NewMetric ¶

func NewMetric() Metric

NewMetric creates a new metric collector with all counters initialized to zero. The returned metric is safe for concurrent use.

func WithAfterFn ¶

func WithAfterFn(afterFn func()) Option

WithAfterFn sets a callback function that will be executed after each job completes. This callback runs regardless of whether the job succeeded or failed. It executes after metrics are updated but before the worker picks up the next task. Useful for cleanup, logging, or triggering post-processing workflows.

Example:

q := NewPool(5, WithAfterFn(func() {
    log.Println("Job completed")
}))

func WithFn ¶

func WithFn(fn func(context.Context, core.TaskMessage) error) Option

WithFn sets a custom handler function that will be called to process tasks. This function is used by the worker's Run method when processing job messages. The context allows cancellation and timeout control during task execution. If not set, defaults to a no-op function that returns nil.

Example:

handler := func(ctx context.Context, msg core.TaskMessage) error {
    // Process the message
    return processTask(msg)
}
q := NewPool(5, WithFn(handler))

func WithLogger ¶

func WithLogger(l Logger) Option

WithLogger sets a custom logger for queue events and errors. By default, the queue uses a standard logger that writes to stderr. Use NewEmptyLogger() to disable logging entirely.

Example:

q := NewPool(5, WithLogger(myCustomLogger))
// or disable logging:
q := NewPool(5, WithLogger(NewEmptyLogger()))

func WithMetric ¶

func WithMetric(m Metric) Option

WithMetric sets a custom metrics collector for tracking queue statistics. The default metric tracks busy workers, success/failure counts, and submitted tasks. Implement the Metric interface to integrate with custom monitoring systems.

Example:

q := NewPool(5, WithMetric(myPrometheusMetric))

func WithObserver ¶

func WithObserver(observer Observer) Option

WithObserver installs a lifecycle observer for queue events.

func WithQueueSize ¶

func WithQueueSize(num int) Option

WithQueueSize sets the maximum capacity of the queue. When set to 0 (default), the queue has unlimited capacity and will grow dynamically. When set to a positive value, Queue() will return ErrMaxCapacity when the limit is reached. Use this to prevent memory exhaustion under high load.

Example:

q := NewPool(5, WithQueueSize(1000)) // Queue will hold at most 1000 pending tasks

func WithRetryInterval ¶

func WithRetryInterval(d time.Duration) Option

WithRetryInterval sets the interval at which the queue polls for new tasks when the queue is empty. This determines how often Request() is retried after receiving ErrNoTaskInQueue. Lower values provide faster response to new tasks but increase CPU usage. Defaults to 1 second.

Example:

q := NewPool(5, WithRetryInterval(100*time.Millisecond)) // Poll every 100ms

func WithWorker ¶

func WithWorker(w core.Worker) Option

WithWorker sets a custom worker implementation for the queue backend. By default, NewPool uses an in-memory Ring buffer worker. Use this to integrate external queue systems like NSQ, NATS, Redis, or RabbitMQ. This option is required when using NewQueue() instead of NewPool().

Example:

q, _ := NewQueue(WithWorker(myNSQWorker), WithWorkerCount(10))

func WithWorkerCount ¶

func WithWorkerCount(num int64) Option

WithWorkerCount sets the number of concurrent worker goroutines that will process jobs. If num is less than or equal to 0, it defaults to runtime.NumCPU(). More workers allow higher concurrency but consume more system resources.

Example:

q := NewPool(10, WithWorkerCount(4)) // Creates a pool with 4 workers

func WithWorkerLifecycle ¶

func WithWorkerLifecycle(lifecycle *management.WorkerLifecycle) Option

WithWorkerLifecycle enables queue-owned management admission, drain, status, desired-state, and lifecycle command handling.

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/faustbrian/go-queue"
	"github.com/faustbrian/go-queue/core"
	"github.com/faustbrian/go-queue/management"
)

func main() {
	now := time.Date(2026, 7, 16, 10, 0, 0, 0, time.UTC)
	lifecycle, err := management.NewWorkerLifecycle(
		management.WorkerLifecycleConfig{
			Metadata: management.StatusMetadata{
				ID: "worker-1", Version: "v1.0.0", Concurrency: 1,
				Protocol: management.ProtocolVersion{Major: 1},
			},
			WorkerGroup: "payments", Queue: "critical",
			MaxCommandResults: 100, Now: func() time.Time { return now },
		},
	)
	if err != nil {
		log.Fatal(err)
	}
	q, err := queue.NewQueue(
		queue.WithWorker(exampleManagedWorker{now: now}),
		queue.WithWorkerLifecycle(lifecycle),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		_ = q.ReleaseContext(ctx)
	}()

	status, err := q.ObserveWorker(context.Background())
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(status.State, status.CurrentJobs)
}

type exampleManagedWorker struct {
	now time.Time
}

func (exampleManagedWorker) Run(context.Context, core.TaskMessage) error { return nil }
func (exampleManagedWorker) Shutdown() error                             { return nil }
func (exampleManagedWorker) Queue(core.TaskMessage) error                { return nil }
func (exampleManagedWorker) Request() (core.TaskMessage, error) {
	return nil, queue.ErrNoTaskInQueue
}
func (w exampleManagedWorker) ObserveWorker(context.Context) (management.WorkerStatus, error) {
	return management.WorkerStatus{
		ID: "worker-1", Version: "v1.0.0", StartedAt: w.now,
		HeartbeatAt: w.now, Queues: []string{"critical"}, Concurrency: 1,
		State: management.WorkerRunning, DrainStatus: management.DrainNotRequested,
		Backend: "example", Protocol: management.ProtocolVersion{Major: 1},
	}, nil
}
func (w exampleManagedWorker) ObserveQueue(context.Context) (management.QueueStatus, error) {
	return management.QueueStatus{
		Backend: "example", Queue: "critical", ObservedAt: w.now,
	}, nil
}

Output:
running 0

func NewOptions ¶

func NewOptions(opts ...Option) *Options

NewOptions creates an Options struct with default values and applies any provided options. Default values:

• workerCount: runtime.NumCPU()
• queueSize: 0 (unlimited)
• logger: stderr logger with timestamps
• worker: nil (must be provided via WithWorker or use NewPool which sets Ring)
• fn: no-op function returning nil
• metric: built-in metric tracker
• retryInterval: 1 second

func NewPool ¶

func NewPool(size int64, opts ...Option) *Queue

NewPool creates a ready-to-use in-memory queue with the Ring buffer worker. This is the recommended way to create a queue for most use cases.

Key differences from NewQueue:

• Automatically creates and attaches a Ring buffer worker (no need for WithWorker)
• Calls Start() automatically so the queue begins processing immediately
• Panics on error instead of returning an error (simplifies initialization)

Parameters:

• size: Number of worker goroutines (if <= 0, defaults to runtime.NumCPU())
• opts: Additional options to customize the queue (WithLogger, WithQueueSize, etc.)

Example:

// Create a pool with 5 workers and custom capacity
q := queue.NewPool(5, queue.WithQueueSize(100))
defer q.Release()

// Queue tasks
q.QueueTask(func(ctx context.Context) error {
    // Process task
    return nil
})

Use NewQueue instead if you need:

• Custom worker implementations (NSQ, NATS, Redis, etc.)
• Manual control over when to start the queue
• Error handling during queue creation

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/faustbrian/go-queue"
)

func main() {
	taskN := 7
	rets := make(chan int, taskN)
	// allocate a pool with 5 goroutines to deal with those tasks
	p := queue.NewPool(5)
	// don't forget to release the pool in the end
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		_ = p.ReleaseContext(ctx)
	}()

	// assign tasks to asynchronous goroutine pool
	for i := 0; i < taskN; i++ {
		idx := i
		if err := p.QueueTask(func(context.Context) error {
			rets <- idx
			return nil
		}); err != nil {
			log.Println(err)
		}
	}

	// wait until all tasks done
	for i := 0; i < taskN; i++ {
		select {
		case index := <-rets:
			fmt.Println("index:", index)
		case <-time.After(time.Second):
			return
		}
	}

}

Output:
index: 3
index: 0
index: 2
index: 4
index: 5
index: 6
index: 1

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/faustbrian/go-queue"
	"github.com/faustbrian/go-queue/job"
)

func main() {
	taskN := 7
	rets := make(chan int, taskN)
	resps := make(chan error, 1)
	completed := make(chan struct{}, taskN)
	// allocate a pool with 5 goroutines to deal with those tasks
	q := queue.NewPool(5, queue.WithAfterFn(func() {
		completed <- struct{}{}
	}))
	// don't forget to release the pool in the end
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		_ = q.ReleaseContext(ctx)
	}()

	// assign tasks to asynchronous goroutine pool
	for i := 0; i < taskN; i++ {
		idx := i
		if err := q.QueueTask(func(ctx context.Context) error {
			// panic job
			if idx == 5 {
				panic("system error")
			}
			// timeout job
			if idx == 6 {
				<-ctx.Done()
			}
			select {
			case <-ctx.Done():
				resps <- ctx.Err()
			default:
			}

			rets <- idx
			return nil
		}, job.AllowOption{
			Timeout: job.Time(100 * time.Millisecond),
		}); err != nil {
			log.Println(err)
		}
	}

	// wait until all tasks done
	for i := 0; i < taskN-1; i++ {
		select {
		case index := <-rets:
			fmt.Println("index:", index)
		case <-time.After(time.Second):
			return
		}
	}
	for i := 0; i < taskN; i++ {
		select {
		case <-completed:
		case <-time.After(time.Second):
			return
		}
	}
	close(resps)
	for e := range resps {
		fmt.Println(e.Error())
	}

	fmt.Println("success task count:", q.SuccessTasks())
	fmt.Println("failure task count:", q.FailureTasks())
	fmt.Println("submitted task count:", q.SubmittedTasks())

}

Output:
index: 3
index: 0
index: 2
index: 4
index: 6
index: 1
context deadline exceeded
success task count: 5
failure task count: 2
submitted task count: 7

func NewQueue ¶

func NewQueue(opts ...Option) (*Queue, error)

NewQueue creates and returns a new Queue instance with the provided options. Returns an error if no worker is specified.

func NewRing ¶

func NewRing(opts ...Option) *Ring

NewRing creates a new Ring instance with the provided options. It initializes the task queue with a default size of 2, sets the capacity based on the provided options, and configures the logger and run function. The function returns a pointer to the newly created Ring instance.

Parameters:

opts - A variadic list of Option functions to configure the Ring instance.

Returns:

*Ring - A pointer to the newly created Ring instance.