queue

package module
v0.0.0-...-e8da5e4 Latest Latest
Warning

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

 Go to latest Published: Aug 28, 2026 License: MIT Imports: 16 Imported by: 0

README

queue

queue is a consolidated worker queue with owned implementations for in-memory, Redis Pub/Sub, Redis Streams, Valkey Streams, NATS, NSQ, and RabbitMQ. It preserves the recognizable golang-queue programming model while owning correctness, operations, and releases in one module.

Status

The package is pre-v1 and undergoing hardening. Production code is held to meaningful 100% coverage; durable delivery claims require backend-specific integration evidence.

Requirements

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

Installation

go get github.com/faustbrian/golib/pkg/queue

Backend packages ship in the same module and are imported explicitly.

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 queueservice. 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, NSQ, and RabbitMQ paths with explicit 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 module and release unit for all maintained backends
  • 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.

AI tools can use llms.txt and llms-full.txt. Release history is maintained in CHANGELOG.md.

Development

Run make check before submitting a change. Backend changes must also pass make integration with the services documented in CONTRIBUTING.md.

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.

Ecosystem

Use the Golib documentation portal to choose companion packages, supported stacks, recipes, and operations guidance.

Documentation

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.
	// This is a terminal state indicating the queue has been shut down and drained.
	// Once this error appears, no new tasks will be processed.
	// Triggered by: calling Shutdown() or Release() on the queue.
	ErrQueueHasBeenClosed = errors.New("golang-queue: queue has been closed")

	// ErrMaxCapacity is returned by Queue() or QueueTask() when the queue is at maximum capacity.
	// This only occurs when WithQueueSize() is used to set a capacity limit.
	// To handle: retry later, drop the task, or process it synchronously.
	// Triggered by: attempting to enqueue when count >= capacity.
	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           EventKind
	Backend        string
	Queue          string
	OccurredAt     time.Time
	Duration       time.Duration
	RetryRemaining int64
	RetryDelay     time.Duration
	Depth          int64
	JobAge         time.Duration
	Err            error
	Classification management.Classification
	FailureCode    string
}

Event describes a queue or backend lifecycle transition.

type EventKind 

type EventKind string

EventKind identifies an observable queue lifecycle transition. 

const (
	EventEnqueued          EventKind = "enqueued"
	EventHandlerStarted    EventKind = "handler_started"
	EventRetryScheduled    EventKind = "retry_scheduled"
	EventHandlerSucceeded  EventKind = "handler_succeeded"
	EventHandlerFailed     EventKind = "handler_failed"
	EventAcknowledged      EventKind = "acknowledged"
	EventAckFailed         EventKind = "ack_failed"
	EventRejected          EventKind = "rejected"
	EventRejectFailed      EventKind = "reject_failed"
	EventShutdownStarted   EventKind = "shutdown_started"
	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)

Implement this interface to integrate with custom logging systems (logrus, zap, etc.).

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()))
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 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

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.
	// This value can range from 0 to the configured workerCount.
	BusyWorkers() int64

	// SuccessTasks returns the total number of tasks that completed successfully.
	// A task is considered successful if it returns no error and doesn't panic.
	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(Event)
}

Observer receives synchronous queue lifecycle events.

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

Example 
package main

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

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

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

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

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

	"github.com/faustbrian/golib/pkg/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
Example (QueueTaskTimeout) 
package main

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

	"github.com/faustbrian/golib/pkg/queue"
	"github.com/faustbrian/golib/pkg/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 (*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 number of workers currently processing jobs.

func (*Queue) CloseAdmission 

func (q *Queue) CloseAdmission() error

CloseAdmission synchronously and idempotently rejects new queue submissions and stops taking new backend work. Already accepted submissions and active handlers retain queue ownership until ReleaseContext drains and releases them.

Example 
package main

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

	"github.com/faustbrian/golib/pkg/queue"
)

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 enqueues a single job (core.QueuedMessage) into the queue. Accepts job options for customization.

func (*Queue) QueueTask 

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

QueueTask enqueues a single task function into the queue. Accepts job options for customization.

func (*Queue) Release 

func (q *Queue) Release()

Release performs a graceful shutdown and waits for all goroutines to finish.

func (*Queue) ReleaseContext 

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

ReleaseContext stops admission, lets already admitted handlers finish, then releases the concrete worker. If ctx expires, the worker remains open so a later call can resume safe release.

func (*Queue) Shutdown 

func (q *Queue) Shutdown()

Shutdown initiates a graceful shutdown of the queue. It signals all goroutines to stop, shuts down the worker, and closes the quit channel. Shutdown is idempotent and safe to call multiple times.

func (*Queue) Start 

func (q *Queue) Start()

Start launches all worker goroutines and begins processing jobs. If workerCount is zero, Start is a no-op.

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 dynamically updates the number of worker goroutines. Triggers scheduling to adjust to the new worker count.

func (*Queue) Wait 

func (q *Queue) Wait()

Wait blocks until all goroutines in the routine group have finished.

func (*Queue) WaitContext 

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

WaitContext waits for all queue-owned routines within ctx.

type Ring 

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

Ring is an in-memory worker implementation using a dynamic circular buffer. It implements the core.Worker interface and provides automatic resizing:

  • Doubles capacity when full
  • Halves capacity when less than 25% utilized

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
  • When head == tail, the queue is empty
  • Both pointers wrap around using modulo arithmetic

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.

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 buffer shrinks automatically (halves in size) when less than 25% full. 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 for all tasks to be processed before completing the shutdown.

Source Files

View all Source files

Directories

Path Synopsis
cmd
semvercheck command
examples
inmemory command
redis command
valkey command
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
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 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 authenticated HTTP boundary.
 Package managementhttp transports queue management contracts over a bounded authenticated HTTP boundary.
Package mocks is a generated GoMock package.
 Package mocks is a generated GoMock package.
queueservice module
Package valkeystream provides a Valkey Streams queue backend.
 Package valkeystream provides a Valkey Streams queue backend.

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.