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

queue

package module
v2.0.0 Latest Latest
Warning

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

 Go to latest Published: Oct 22, 2026 License: MIT Imports: 16 Imported by: 0

README

queue

CI CodeQL Documentation Go Reference Release Go License

queue is a consolidated worker queue with root-module implementations for in-memory, Redis Pub/Sub, Redis Streams, Valkey Streams, NATS, and NSQ. Independently released target-oriented adapters provide RabbitMQ and service lifecycle integration. The RabbitMQ adapter delegates AMQP policy to go-rabbitmq-queues.

Status

The root module is stable at v2; the independently released adapters keep their own versions. Version 2 removed the former redisstream, rabbitmq and queueservice compatibility facades; see the migration guide. Durable delivery claims require backend-specific integration evidence.

Requirements

  • Go 1.27.0 or later
  • a supported broker for non-memory backends

CI and the development workspace use Go 1.27.3. Build applications with a patched Go release; this CI selection does not raise module language minimums or patch the runtime of already-built applications.

Installation

go get github.com/faustbrian/go-queue/v2@v2.0.0

Root backend packages, including adapters/redisstream, ship in the same module and are imported explicitly. Install the independently versioned github.com/faustbrian/go-queue/adapters/rabbitmq or github.com/faustbrian/go-queue/adapters/service module when needed. Both still build on root v1 until their root-v2 majors are published; keep their users on root v1 until then.

Choosing this package

Use queue for bounded worker coordination with explicit backend delivery, retry, settlement, and shutdown semantics. Use the in-memory backend for local work, Pub/Sub or Core NATS for disposable delivery, and a durable backend only after checking the backend matrix.

Applications own handlers, idempotency, configuration, broker deployment, and scheduling. The separate go-queue-control-plane owns administrative authorization, audit, and policy; this library exposes worker management contracts, not an operator dashboard. For RabbitMQ-native messaging use go-rabbitmq-queues; RabbitMQ Streams use the separate go-rabbitmq-streams, not this AMQP job adapter. No backend provides exactly-once application effects.

Package and module map

Import below github.com/faustbrian/go-queue Responsibility Release ownership
root, core, job Coordination, worker contracts, job envelopes Root module
redisdb, adapters/redisstream, valkeystream, nats, nsq Explicit native backends Root module
management, managementhttp Capability-negotiated worker status and control Root module
adapters/rabbitmq AMQP job integration over go-rabbitmq-queues Independent nested module
adapters/service go-service admission, drain, and shutdown integration Independent nested module

Select and install only the modules needed. See the release guide for nested-module tags and compatibility guidance before an upgrade.

Quickstart

worker, err := redisdb.NewWorkerE(
    redisdb.WithAddr("127.0.0.1:6379"),
    redisdb.WithChannel("jobs"),
    redisdb.WithRunFunc(func(ctx context.Context, task core.TaskMessage) error {
        return handle(ctx, task.Payload())
    }),
)
if err != nil {
    return err
}

q, err := queue.NewQueue(queue.WithWorker(worker), queue.WithWorkerCount(8))
if err != nil {
    return err
}
q.Start()
defer q.Release()

Redis Pub/Sub is low-latency and non-durable. Use Redis Streams or Valkey Streams when work must remain pending until settlement. They are independent native backends; adopting Valkey does not require removing Redis. Read delivery semantics before selecting a backend. Scheduler and API processes that only submit Valkey work should use valkeystream.NewPublisherE; it appends jobs without joining a consumer group or starting worker loops.

Services should compose concrete producers and workers through adapters/service. The adapter keeps concrete queue APIs visible, closes queue admission during service drain, drains accepted publishers before closing an owned transport, uses the existing correlation queue boundary for every message and delivery attempt, and optionally propagates bounded W3C trace context through an explicit caller-owned OpenTelemetry propagator.

External control planes should depend on the backend-neutral contracts in management. Incompatible workers remain visible, but management capabilities are enabled only when both peers report support. The managementhttp package makes those contracts remotely callable without exposing backend clients. managementhttp.NewFleetClient can resolve a changing set of worker endpoints for multi-replica deployments, aggregate their status, route worker-specific commands, and fan queue or worker-group lifecycle commands to every current replica.

Package Guarantees

  • explicit retry, acknowledgement, redelivery, cancellation, and shutdown behavior
  • safe failure classification and codes that preserve errors.Is while redacting arbitrary handler, panic, and settlement text
  • handler backoff limited to retryable failures so terminal and uncertain classifications reach backend settlement without repeated side effects
  • optional one-time decoded-delivery validation before handler retry execution
  • durable Redis Streams, Valkey Streams, and NSQ paths with explicit settlement
  • an independently versioned RabbitMQ adapter with confirmed publish-before-ack settlement
  • observable lifecycle events, metrics, and backend identity
  • stable management-protocol version and capability negotiation for external control planes
  • bounded worker and queue status contracts that distinguish unsupported backend measurements from measured zero values, with paginated readers
  • bounded authenticated HTTP transport for remote status, records, and control commands
  • bounded dynamic management fleets with fail-closed discovery, complete status aggregation, worker routing, and explicit partial fan-out outcomes
  • backend-neutral command enforcement contracts with explicit confirmation, bounded bulk retry, acknowledgement, timeout, partial, and unknown outcomes
  • revisioned desired-state reconciliation with monotonic per-target application, retry-safe failures, and caller-owned scheduling
  • queue-owned pause, resume, drain, and terminate enforcement that stops admission at safe boundaries and reports in-flight work honestly
  • bounded failed-job and dead-letter inspection with payloads hidden by default and privileged content capped at one mebibyte
  • one root release unit for maintained root backends, with explicit nested release units for optional integrations
  • backend-specific guarantees documented without abstraction leakage

Documentation

Start with the documentation index, quickstart, adoption guide, and API reference. Review the backend matrix, failure model, service integration, and integration evidence before production use. Valkey adopters should use the Valkey 9 Streams guide and runnable example. RabbitMQ migrations should use the target-oriented adapter guide. Release history is maintained in CHANGELOG.md.

For ecosystem-wide selection and ownership guidance, see the versioned Golib ecosystem index and Persistence and durability family guidance.

Development

Use GOWORK=off go test ./... -count=1 -timeout=3m for standalone untagged tests. make check and make ci are bounded, public-verifier routes with the manifest's real service and integration-tag requirements; make check-full retains explicit full qualification. See CONTRIBUTING.md for focused active fuzzing, lifecycle races, and backend fixtures.

Contributing

Read CONTRIBUTING.md and follow the code of conduct. Every backend change must document its delivery and settlement impact.

Security

Report vulnerabilities privately according to SECURITY.md. Review docs/security.md before processing untrusted jobs.

License

queue is available under the MIT License. Fork provenance and third-party attribution are recorded in NOTICE and THIRD_PARTY_NOTICES.md.

Documentation

Overview

Package queue schedules in-process tasks and backend deliveries with bounded worker concurrency and per-delivery retries. Backend acceptance, settlement and durability are distinct from handler completion; none promises exactly-once application effects. Handlers must cooperate with their contexts.

Construct queues with NewQueue or NewPool. CloseAdmission stops new submissions without closing the worker; ReleaseContext drains tracked scheduling/work wrappers before worker shutdown. A timed-out handler can outlive those wrappers when it ignores cancellation. Message payload bytes are not copied on admission.

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	// ErrNoTaskInQueue is returned by Worker.Request() when the queue is currently empty.
	// This is a temporary condition - new tasks may be added later.
	// The queue scheduler uses this error to determine when to retry or wait for notifications.
	ErrNoTaskInQueue = errors.New("golang-queue: no task in queue")

	// ErrQueueHasBeenClosed is returned by Worker.Request() during/after shutdown when no tasks remain.
	// It describes intake closure, not proof that handlers or side effects finished.
	ErrQueueHasBeenClosed = errors.New("golang-queue: queue has been closed")

	// ErrMaxCapacity reports rejected admission to a full Ring or bounded stream.
	// The default Ring is bounded; backend stream capacity is configured separately.
	ErrMaxCapacity = errors.New("golang-queue: maximum size limit reached")

	// ErrInvalidConfiguration reports unsafe queue scheduler configuration.
	ErrInvalidConfiguration = errors.New("golang-queue: invalid configuration")

	// ErrHandlerPanic reports a recovered handler panic without exposing the
	// arbitrary panic value across logs, observations, or durable settlement.
	ErrHandlerPanic = errors.New("golang-queue: handler panic")

	// ErrSettlementPanic reports a recovered backend settlement callback panic
	// without exposing its arbitrary panic value.
	ErrSettlementPanic = errors.New("golang-queue: settlement panic")

	// ErrWorkerShutdownPanic reports a recovered concrete worker shutdown panic
	// without exposing or retaining its arbitrary panic value.
	ErrWorkerShutdownPanic = errors.New("golang-queue: worker shutdown panic")
)
View Source 
var (
	// ErrInvalidManagementLifecycle reports a lifecycle paired with a worker
	// that cannot provide native worker and queue status.
	ErrInvalidManagementLifecycle = errors.New("queue: invalid management lifecycle")
	// ErrManagementLifecycleDisabled reports management calls on a queue that
	// was not explicitly configured for them.
	ErrManagementLifecycleDisabled = errors.New("queue: management lifecycle disabled")
)
View Source 
var ErrMissingWorker = errors.New("missing worker module")

ErrMissingWorker is returned when a queue is created without a worker implementation.

View Source 
var ErrQueueShutdown = errors.New("queue has been closed and released")

ErrQueueShutdown is returned when an operation is attempted on a queue that has already been closed and released.

Functions

This section is empty.

Types

type Event 

type Event struct {
	// Kind selects the observed transition.
	Kind EventKind
	// Backend is a best-effort backend label, not authenticated identity.
	Backend string
	// Queue is a best-effort logical queue label; Ring leaves it empty.
	Queue string
	// OccurredAt is local event time, not a distributed ordering clock.
	OccurredAt time.Time
	// Duration is measured elapsed work when supplied; zero can mean unavailable.
	Duration time.Duration
	// RetryRemaining counts remaining additional handler attempts, not broker deliveries.
	RetryRemaining int64
	// RetryDelay reports the scheduled local retry delay.
	RetryDelay time.Duration
	// Depth reports Ring depth for enqueue; other events may leave zero.
	Depth int64
	// JobAge is best-effort age when supplied, or zero when unavailable.
	JobAge time.Duration
	// Err retains the classifiable error, which may include sensitive caller text.
	Err error
	// Classification is the selected backend-neutral disposition.
	Classification management.Classification
	// FailureCode is an operational label, not arbitrary error text.
	FailureCode string
}

Event describes a queue or backend lifecycle transition.

type EventKind 

type EventKind string

EventKind identifies an observable queue lifecycle transition. 

const (
	// EventEnqueued reports successful submission, not handler completion.
	EventEnqueued EventKind = "enqueued"
	// EventHandlerStarted reports the start of a tracked work wrapper.
	EventHandlerStarted EventKind = "handler_started"
	// EventRetryScheduled reports a local handler retry delay.
	EventRetryScheduled EventKind = "retry_scheduled"
	// EventHandlerSucceeded reports success after handler and required settlement.
	EventHandlerSucceeded EventKind = "handler_succeeded"
	// EventHandlerFailed reports final work/settlement failure.
	EventHandlerFailed EventKind = "handler_failed"
	// EventAcknowledged reports successful explicit success settlement.
	EventAcknowledged EventKind = "acknowledged"
	// EventAckFailed reports failed success settlement with possible ambiguous effects.
	EventAckFailed EventKind = "ack_failed"
	// EventRejected reports successful explicit failure settlement.
	EventRejected EventKind = "rejected"
	// EventRejectFailed reports failed failure settlement.
	EventRejectFailed EventKind = "reject_failed"
	// EventShutdownStarted reports entry into shutdown.
	EventShutdownStarted EventKind = "shutdown_started"
	// EventShutdownCompleted reports worker shutdown return, before Wait necessarily completes.
	EventShutdownCompleted EventKind = "shutdown_completed"
)

type Logger 

type Logger interface {
	// Infof logs formatted informational messages.
	Infof(format string, args ...any)

	// Errorf logs formatted error messages.
	Errorf(format string, args ...any)

	// Fatalf logs formatted fatal errors with stack trace information.
	// Used for panics and critical failures.
	Fatalf(format string, args ...any)

	// Info logs informational messages.
	Info(args ...any)

	// Error logs error messages.
	Error(args ...any)

	// Fatal logs fatal errors with stack trace information.
	// Used for panics and critical failures.
	Fatal(args ...any)
}

Logger defines the interface for logging queue events, errors, and fatal conditions. The queue uses this interface to report:

  • Info: Normal operations (shutdown, retry attempts)
  • Error: Recoverable errors (task failures, runtime errors)
  • Fatal: Panics and critical failures (includes stack traces)

Methods can run concurrently and synchronously; implementations must return promptly and coordinate mutable state. Root Queue contains logger panics, but backend-specific direct logging calls may not use that panic boundary.

func NewEmptyLogger 

func NewEmptyLogger() Logger

NewEmptyLogger returns a logger that discards every level without formatting arguments or capturing source-backed stacks.

Example 
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 writes timestamped levels to stderr. Fatal/Fatalf add source-backed stack text but do not exit or panic. Source and formatted arguments are not redacted; callers must control diagnostic content and disclosure.

type Metric 

type Metric interface {
	// IncBusyWorker increments the count of workers currently processing tasks.
	// Called atomically when a worker starts processing a job.
	IncBusyWorker()

	// DecBusyWorker decrements the count of workers currently processing tasks.
	// Called atomically when a worker finishes processing a job (success or failure).
	DecBusyWorker()

	// BusyWorkers returns the current number of workers actively processing tasks.
	// Root Queue counts work wrappers, not canceled handlers still running.
	// Direct collector calls are not range-checked.
	BusyWorkers() int64

	// SuccessTasks returns the total number of tasks that completed successfully.
	// Root Queue records success only after handler and any required settlement
	// return without error. This is process-local accounting, not broker history.
	SuccessTasks() uint64

	// FailureTasks returns the total number of tasks that failed.
	// A task is considered failed if it returns an error, panics, or times out.
	FailureTasks() uint64

	// SubmittedTasks returns the total number of tasks submitted to the queue.
	// This includes tasks still pending, in progress, and completed.
	SubmittedTasks() uint64

	// CompletedTasks returns the total number of tasks that have finished processing.
	// This equals SuccessTasks() + FailureTasks().
	CompletedTasks() uint64

	// IncSuccessTask increments the successful task counter.
	// Called atomically after a task completes without error.
	IncSuccessTask()

	// IncFailureTask increments the failed task counter.
	// Called atomically after a task fails, panics, or times out.
	IncFailureTask()

	// IncSubmittedTask increments the submitted task counter.
	// Called atomically when a new task is queued.
	IncSubmittedTask()
}

Metric defines the interface for tracking queue performance and worker statistics. All methods must be safe for concurrent access from multiple goroutines. Implement this interface to integrate with custom monitoring systems (Prometheus, StatsD, etc.).

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.

type Observer 

type Observer interface {
	// Observe receives synchronous events, potentially concurrently from multiple workers.
	Observe(Event)
}

Observer receives synchronous, potentially concurrent queue events. Queue contains observer panics, but cannot preempt a blocking observer or redact caller errors. Observers must return promptly and control their own logging.

type ObserverFunc 

type ObserverFunc func(Event)

ObserverFunc adapts a function to Observer.

func (ObserverFunc) Observe 

func (f ObserverFunc) Observe(event Event)

Observe calls the wrapped observer function.

type Option 

type Option interface {
	// contains filtered or unexported methods
}

Option is a functional option for configuring a Queue. It follows the functional options pattern for flexible and extensible configuration.

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 runs synchronously after metrics/observation. Scheduling is already released, so another task can start first. Calls may overlap, and a blocking callback delays its work wrapper; panics are contained by Root Queue. 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. It configures Ring.Run; custom/network workers own their own handler options. 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. The Ring defaults to 10,000 pending jobs; zero explicitly makes it unbounded. This does not configure capacity for external workers or count running jobs. 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 delays polling after Request errors, including but not limited to empty intake. Notifications/admission closure can wake the wait earlier. 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.

Example 
package main

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

	"github.com/faustbrian/go-queue/v2"
	"github.com/faustbrian/go-queue/v2/core"
	"github.com/faustbrian/go-queue/v2/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

type OptionFunc 

type OptionFunc func(*Options)

OptionFunc is a function adapter that implements the Option interface. It allows regular functions to be used as Options.

type Options 

type Options struct {
	// contains filtered or unexported fields
}

Options holds the configuration parameters for a Queue. Use the With* functions to configure these options when creating a queue.

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: 10,000 pending Ring jobs (zero explicitly disables the bound)
  • 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

type Queue 

type Queue struct {
	sync.Mutex // Mutex to protect concurrent access to queue state
	// contains filtered or unexported fields
}

Queue represents a message queue with worker management, job scheduling, retry logic, and graceful shutdown capabilities.

func NewPool 

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

NewPool constructs a Ring-backed Queue, starts it and panics on configuration error. Nonpositive size selects runtime.NumCPU. Options are applied after the initial worker/count choices and can override them, including WithWorker. Release the returned queue when finished.

Example (QueueTask) 
package main

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

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

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
Example (QueueTaskTimeout) 
package main

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

	"github.com/faustbrian/go-queue/v2"
	"github.com/faustbrian/go-queue/v2/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 validates scheduler options and retains the supplied worker, logger, metric, observer and lifecycle. It does not start the scheduler or assume a worker has no constructor-owned background work. Use Start and release the queue when finished. A typed-nil Worker is not detected by the interface check.

func (*Queue) ApplyDesiredState 

func (q *Queue) ApplyDesiredState(
	ctx context.Context,
	record management.DesiredRecord,
) error

ApplyDesiredState converges queue admission and graceful shutdown to one durable revision.

func (*Queue) BusyWorkers 

func (q *Queue) BusyWorkers() int64

BusyWorkers returns the internal active work-wrapper count, independent of custom metrics. Timed-out handlers still running are not counted.

func (*Queue) CloseAdmission 

func (q *Queue) CloseAdmission() error

CloseAdmission synchronously and idempotently rejects new queue submissions and withdraws new backend intake. An already accepted submission/request or work wrapper can remain active. It does not cancel handlers, close transports or join a handler that ignores its own timeout.

Example 
package main

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

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

func main() {
	q := queue.NewPool(1)
	if err := q.CloseAdmission(); err != nil {
		log.Fatal(err)
	}

	err := q.QueueTask(func(context.Context) error { return nil })
	fmt.Println(errors.Is(err, queue.ErrQueueShutdown))

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err = q.ReleaseContext(ctx); err != nil {
		log.Fatal(err)
	}
}

Output:
true

func (*Queue) CompletedTasks 

func (q *Queue) CompletedTasks() uint64

CompletedTasks returns the total number of completed tasks (success + failure).

func (*Queue) Execute 

func (q *Queue) Execute(
	ctx context.Context,
	command management.Command,
) (management.CommandResult, error)

Execute applies one queue-owned lifecycle command. Backend record and queue mutation actions remain explicitly unsupported by WorkerLifecycle.

func (*Queue) FailureTasks 

func (q *Queue) FailureTasks() uint64

FailureTasks returns the number of failed tasks.

func (*Queue) ObserveQueue 

func (q *Queue) ObserveQueue(ctx context.Context) (management.QueueStatus, error)

ObserveQueue forwards honest backend-native queue measurements.

func (*Queue) ObserveWorker 

func (q *Queue) ObserveWorker(ctx context.Context) (management.WorkerStatus, error)

ObserveWorker overlays queue-owned admission and in-flight state on the backend worker's native status.

func (*Queue) Queue 

func (q *Queue) Queue(message core.QueuedMessage, opts ...job.AllowOption) error

Queue borrows message.Bytes(), applies only the first job option, validates execution metadata and synchronously invokes Worker.Queue. Success is backend admission, not completed execution. There is no caller context; a blocked backend publish also holds an admission until it returns.

func (*Queue) QueueTask 

func (q *Queue) QueueTask(task job.TaskFunc, opts ...job.AllowOption) error

QueueTask admits an in-process callback using only the first job option. Task functions are not serialized, so use a Ring for actual local execution. The callback must honor the message's total timeout; admission does not copy application-owned state captured by it.

func (*Queue) Release 

func (q *Queue) Release()

Release starts an unstarted Ring-backed queue, calls Shutdown and waits for the scheduler/work wrappers without a caller deadline. Timeout-ignoring handler goroutines are not joined; this is not a graceful drain-before-close guarantee.

func (*Queue) ReleaseContext 

func (q *Queue) ReleaseContext(ctx context.Context) error

ReleaseContext closes admission, waits for accepted publishes and tracked scheduler/work wrappers, then shuts down the worker. The context bounds drain waits, not synchronous Worker.Shutdown or noncooperative callbacks. Expiration before shutdown leaves the worker open for a later release attempt. A handler that ignores its own timeout can continue after its tracked wrapper completes. A nil context returns context.Canceled; shutdown errors remain cached.

func (*Queue) Shutdown 

func (q *Queue) Shutdown()

Shutdown rejects submissions, calls the concrete worker's synchronous Shutdown, then signals tracked work to cancel. It is idempotent but does not join work or bound Worker.Shutdown. Ring shutdown waits for buffer dequeue. Use ReleaseContext for the separate drain-before-close path.

func (*Queue) Start 

func (q *Queue) Start()

Start launches the intake scheduler once. Work goroutines are created as capacity becomes available. It does nothing after shutdown or when the current worker count is zero; constructor options normally select runtime.NumCPU.

func (*Queue) SubmittedTasks 

func (q *Queue) SubmittedTasks() uint64

SubmittedTasks returns the number of tasks submitted to the queue.

func (*Queue) SuccessTasks 

func (q *Queue) SuccessTasks() uint64

SuccessTasks returns the number of successfully completed tasks.

func (*Queue) UpdateWorkerCount 

func (q *Queue) UpdateWorkerCount(num int64)

UpdateWorkerCount changes the admission concurrency threshold and schedules work. It does not stop existing handlers. Unlike WithWorkerCount, zero/negative values are stored as-is and prevent new work from being scheduled.

func (*Queue) Wait 

func (q *Queue) Wait()

Wait joins the current scheduler/work-wrapper group; it does not close intake, join canceled inner handler goroutines or prevent later work from being added.

func (*Queue) WaitContext 

func (q *Queue) WaitContext(ctx context.Context) error

WaitContext bounds the current scheduler/work-wrapper wait. It does not cancel work or join timeout-ignoring handlers; a nil context returns context.Canceled.

type Ring 

type Ring struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Ring is a mutex-protected FIFO of borrowed task references, constructed with NewRing. Its zero value is not usable. Queued messages are neither copied nor durable, and must not be mutated concurrently by the caller. It implements the core.Worker interface and provides automatic resizing:

  • Doubles capacity when full
  • Halves backing storage at or below half occupancy, down to two slots

The ring buffer uses two pointers (head and tail) to track the queue boundaries:

  • head: points to the next task to dequeue
  • tail: points to the next empty slot for enqueuing
  • count distinguishes empty and full when head == tail
  • Both pointers wrap around using modulo arithmetic

func NewRing 

func NewRing(opts ...Option) *Ring

NewRing allocates a two-slot buffer with a 10,000 pending-job admission bound unless WithQueueSize changes it. Zero capacity is unbounded; negative capacity panics. WithFn supplies synchronous task execution; it must honor its context.

func (*Ring) BackendName 

func (*Ring) BackendName() string

BackendName identifies the in-memory worker in lifecycle events.

func (*Ring) Queue 

func (s *Ring) Queue(task core.TaskMessage) error

Queue adds a task to the ring buffer. The buffer grows dynamically (doubles in size) when full, unless capacity is set. Returns ErrQueueShutdown if the queue is closing, or ErrMaxCapacity if at the size limit.

Thread-safety: This method is safe for concurrent calls.

func (*Ring) QueueName 

func (*Ring) QueueName() string

QueueName is empty because an in-memory ring has no broker queue name.

func (*Ring) Request 

func (s *Ring) Request() (core.TaskMessage, error)

Request dequeues and returns the next task from the ring buffer. The backing buffer halves at or below half occupancy, down to two slots. Returns:

  • (task, nil) if a task is successfully dequeued
  • (nil, ErrNoTaskInQueue) if the queue is currently empty
  • (nil, ErrQueueHasBeenClosed) if shutdown is complete and the queue is empty

During shutdown, this method signals the exit channel when the last task is dequeued, allowing Shutdown() to complete.

Thread-safety: This method is safe for concurrent calls.

func (*Ring) Run 

func (s *Ring) Run(ctx context.Context, task core.TaskMessage) error

Run executes a new task using the provided context and task message. It calls the runFunc function, which is responsible for processing the task. The context allows for cancellation and timeout control of the task execution.

func (*Ring) Shutdown 

func (s *Ring) Shutdown() error

Shutdown gracefully shuts down the worker. It sets the stopFlag to indicate that the queue is shutting down and prevents new tasks from being added. If the queue is already shut down, it returns ErrQueueShutdown. It waits until the buffer is dequeued, not until handlers or settlement finish. A consumer must keep requesting backlog for this wait to complete.

Source Files

View all Source files

Directories

Path Synopsis
adapters
redisstream
Package redisstream owns Redis Streams intake, stale pending-entry reclaim and explicit settlement.
 Package redisstream owns Redis Streams intake, stale pending-entry reclaim and explicit settlement.
cmd
semvercheck command
Command semvercheck validates one v-prefixed tag or prints the next stable patch, minor or major version.
 Command semvercheck validates one v-prefixed tag or prints the next stable patch, minor or major version.
Package core defines backend worker, message and optional settlement seams.
 Package core defines backend worker, message and optional settlement seams.
examples
inmemory command
Command inmemory demonstrates local Ring admission and process-owned cleanup.
 Command inmemory demonstrates local Ring admission and process-owned cleanup.
redis command
Command redis demonstrates a transient Redis Pub/Sub worker using REDIS_ADDR.
 Command redis demonstrates a transient Redis Pub/Sub worker using REDIS_ADDR.
valkey command
Command valkey demonstrates separately owned Valkey producer and worker connections.
 Command valkey demonstrates separately owned Valkey producer and worker connections.
internal
safeerr
Package safeerr provides errors that preserve a cause without exposing its potentially credential-bearing text through Error.
 Package safeerr provides errors that preserve a cause without exposing its potentially credential-bearing text through Error.
semver
Package semver checks v-prefixed release tag syntax and computes stable version successors for the release command.
 Package semver checks v-prefixed release tag syntax and computes stable version successors for the release command.
streamqueue
Package streamqueue defines package-owned stream queue semantics shared by native backend adapters.
 Package streamqueue defines package-owned stream queue semantics shared by native backend adapters.
testutil/apiguard
Package apiguard provides source-level assertions for public Go APIs.
 Package apiguard provides source-level assertions for public Go APIs.
testutil/streamconformance
Package streamconformance defines the shared behavioral contract exercised by every first-class Streams worker.
 Package streamconformance defines the shared behavioral contract exercised by every first-class Streams worker.
Package job defines mutable execution envelopes encoded with encoding/json.
 Package job defines mutable execution envelopes encoded with encoding/json.
Package management defines stable worker/control-plane contracts without implementing queue delivery or backend operations.
 Package management defines stable worker/control-plane contracts without implementing queue delivery or backend operations.
Package managementhttp transports queue management contracts over a bounded bearer-authenticated HTTP boundary.
 Package managementhttp transports queue management contracts over a bounded bearer-authenticated HTTP boundary.
Package mocks is a generated GoMock package.
 Package mocks is a generated GoMock package.
Package nats adapts Core NATS queue subscriptions, not JetStream.
 Package nats adapts Core NATS queue subscriptions, not JetStream.
Package nsq adapts NSQ topic/channel delivery with manual FIN/REQ settlement.
 Package nsq adapts NSQ topic/channel delivery with manual FIN/REQ settlement.
Package redisdb adapts transient Redis Pub/Sub, not Redis Streams.
 Package redisdb adapts transient Redis Pub/Sub, not Redis Streams.
Package valkeystream owns standalone Valkey Streams producer/consumer clients with bounded admission and stale pending-entry reclaim.
 Package valkeystream owns standalone Valkey Streams producer/consumer clients with bounded admission and stale pending-entry reclaim.

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.