queue

package module

Go to main page

Versions in this module

v0

v0.3.0

Nov 10, 2026

Changes in this version

+ const DriverJobMetadataVersion

+ var ErrOverlapping = errors.New("bus job overlap prevented")

+ var ErrRateLimited = errors.New("bus job rate limited")

+ var ErrSkipped = errors.New("bus job skipped by middleware")

+ var ErrWorkflowUnsupported = errors.New("portable workflows are unsupported")

+ func DriverUniqueKey(job Job, queueName string) string

+ func IsPermanent(err error) bool

+ func Permanent(err error) error

+ type BatchJob struct

+ Job StoredJob

+ JobID string

+ type BatchJobOutcome string

+ const BatchJobFailed

+ const BatchJobSucceeded

+ type BatchRecord struct

+ AllowFailed bool

+ BatchID string

+ CreatedAt time.Time

+ DispatchID string

+ Jobs []BatchJob

+ Name string

+ Queue string

type BatchState

+ AllowFailed bool

+ BatchID string

+ Cancelled bool

+ Completed bool

+ CreatedAt time.Time

+ DispatchID string

+ Failed int

+ Name string

+ Pending int

+ Processed int

+ Queue string

+ Total int

+ UpdatedAt time.Time

+ type ChainNode struct

+ Job StoredJob

+ NodeID string

+ type ChainRecord struct

+ ChainID string

+ CreatedAt time.Time

+ DispatchID string

+ Nodes []ChainNode

+ Queue string

type ChainState

+ ChainID string

+ Completed bool

+ CreatedAt time.Time

+ DispatchID string

+ Failed bool

+ Failure string

+ NextIndex int

+ Nodes []ChainNode

+ Queue string

+ UpdatedAt time.Time

+ type Continuation = busruntime.Continuation

+ type ContinuationBuilder struct

+ func Continue(builder any) *ContinuationBuilder

+ func (b *ContinuationBuilder) Dispatch(ctx context.Context) (string, error)

+ func (b *ContinuationBuilder) OnCompletion(job Job) *ContinuationBuilder

+ func (b *ContinuationBuilder) OnFailure(job Job) *ContinuationBuilder

+ func (b *ContinuationBuilder) OnProgress(job Job) *ContinuationBuilder

+ func (b *ContinuationBuilder) OnSuccess(job Job) *ContinuationBuilder

+ type ContinuationDefinition struct

+ Job StoredJob

+ Phase string

+ type ContinuationOutcome = busruntime.ContinuationOutcome

+ type ContinuationRecord struct

+ Definitions []ContinuationDefinition

+ WorkflowID string

+ type ContinuationSummary = busruntime.ContinuationSummary

type DatabaseConfig

+ DisableAutoMigrate bool

type DispatchResult

+ DispatchID string

type Driver

+ const DriverTemporal

+ type DriverJobMetadata = busruntime.DeliveryMetadata

+ func DriverMetadata(job Job) DriverJobMetadata

type Event

+ BatchID string

+ ChainID string

+ DispatchID string

+ EventID string

+ JobID string

+ Layer EventLayer

+ SchemaVersion int

type EventKind

+ const EventBatchCancelled

+ const EventBatchCompleted

+ const EventBatchFailed

+ const EventBatchProgressed

+ const EventBatchStarted

+ const EventCallbackFailed

+ const EventCallbackStarted

+ const EventCallbackSucceeded

+ const EventChainAdvanced

+ const EventChainCompleted

+ const EventChainFailed

+ const EventChainStarted

+ const EventJobFailed

+ const EventJobStarted

+ const EventJobSucceeded

+ const EventSettlementFailed

+ type EventLayer string

+ const EventLayerQueue

+ const EventLayerWorker

+ const EventLayerWorkflow

type FailOnError

+ When func(err error) bool

+ func (f FailOnError) Handle(ctx context.Context, message Message, next Next) error

type FakeQueue

+ func (f *FakeQueue) AssertBatchCount(t testing.TB, expected int)

+ func (f *FakeQueue) AssertBatched(t testing.TB, predicate func(BatchRecord) bool)

+ func (f *FakeQueue) AssertChained(t testing.TB, expected []string)

+ func (f *FakeQueue) AssertNothingBatched(t testing.TB)

+ func (f *FakeQueue) Batch(jobs ...Job) BatchBuilder

+ func (f *FakeQueue) BatchRecords() []BatchRecord

+ func (f *FakeQueue) BusDispatchDirect(ctx context.Context, jobType string, payload []byte, ...) error

+ func (f *FakeQueue) Chain(jobs ...Job) ChainBuilder

+ func (f *FakeQueue) ChainRecords() []ChainRecord

+ func (f *FakeQueue) ContinuationRecords() []ContinuationRecord

+ func (f *FakeQueue) FindBatch(ctx context.Context, batchID string) (BatchState, error)

+ func (f *FakeQueue) FindChain(ctx context.Context, chainID string) (ChainState, error)

+ func (f *FakeQueue) FindContinuations(ctx context.Context, id string) ([]ContinuationOutcome, error)

+ func (f *FakeQueue) Prune(ctx context.Context, before time.Time) error

type Job

+ func DriverWithMetadata(job Job, metadata DriverJobMetadata) Job

+ func (t Job) PayloadAs[T any]() (T, error)

type Lock

+ Release func(ctx context.Context) error

type Locker

+ Acquire func(ctx context.Context, key string, ttl time.Duration) (Lock, bool, error)

type Message

+ Attempt int

+ BatchID string

+ ChainID string

+ DispatchID string

+ JobID string

+ JobType string

+ SchemaVersion int

+ func NewMessage(jobType string, payload []byte) Message

+ func (m Message) Bind(dst any) error

+ func (m Message) Continuation() (Continuation, bool)

+ func (m Message) PayloadAs[T any]() (T, error)

+ func (m Message) PayloadBytes() []byte

type Middleware

+ Handle func(ctx context.Context, message Message, next Next) error

type MiddlewareFunc

+ func (f MiddlewareFunc) Handle(ctx context.Context, message Message, next Next) error

+ type ObservedJobMetadata struct

+ BatchID string

+ ChainID string

+ DispatchID string

+ JobID string

+ JobKey string

+ JobType string

+ func ResolveObservedJobMetadata(rawType string, payload []byte) ObservedJobMetadata

+ func ResolveObservedJobMetadataFromJob(job Job) ObservedJobMetadata

type Option

+ func WithLegacyDirectEnvelope() Option

+ func WithResourceID(id string) Option

type Queue

+ func (r *Queue) FindContinuations(ctx context.Context, id string) ([]ContinuationOutcome, error)

type RateLimit

+ Key func(ctx context.Context, message Message) string

+ Limiter RateLimiter

+ func (r RateLimit) Handle(ctx context.Context, message Message, next Next) error

type RateLimiter

+ Allow func(ctx context.Context, key string) (allowed bool, retryAfter time.Duration, err error)

type RetryPolicy

+ func (RetryPolicy) Handle(ctx context.Context, message Message, next Next) error

+ type SQLStoreConfig struct

+ AutoMigrate bool

+ DB *sql.DB

+ DSN string

+ DriverName string

type SkipWhen

+ Predicate func(ctx context.Context, message Message) bool

+ func (s SkipWhen) Handle(ctx context.Context, message Message, next Next) error

+ type StoredJob struct

+ Options StoredJobOptions

+ Payload []byte

+ Type string

+ type StoredJobOptions struct

+ Backoff time.Duration

+ Delay time.Duration

+ Queue string

+ Retry int

+ Timeout time.Duration

+ UniqueFor time.Duration

type WithoutOverlapping

+ Key func(ctx context.Context, message Message) string

+ Locker Locker

+ TTL time.Duration

+ func (w WithoutOverlapping) Handle(ctx context.Context, message Message, next Next) error

+ type WorkflowOutcomeStore interface

+ FailChainNode func(ctx context.Context, chainID, nodeID string, cause error) (state ChainState, owned bool, err error)

+ SettleBatchJob func(ctx context.Context, batchID, jobID string, outcome BatchJobOutcome, ...) (state BatchState, owned bool, err error)

type WorkflowStore

+ func NewMemoryStore() WorkflowStore

+ func NewSQLStore(config SQLStoreConfig) (WorkflowStore, error)

+ func NewSQLStoreWithManagedSchema(config SQLStoreConfig) (WorkflowStore, error)

+ AdvanceChain func(ctx context.Context, chainID string, completedNode string) (next *ChainNode, done bool, err error)

+ CancelBatch func(ctx context.Context, batchID string) error

+ CreateBatch func(ctx context.Context, rec BatchRecord) error

+ CreateChain func(ctx context.Context, rec ChainRecord) error

+ FailChain func(ctx context.Context, chainID string, cause error) error

+ GetBatch func(ctx context.Context, batchID string) (BatchState, error)

+ GetChain func(ctx context.Context, chainID string) (ChainState, error)

+ MarkBatchJobFailed func(ctx context.Context, batchID, jobID string, cause error) (BatchState, bool, error)

+ MarkBatchJobStarted func(ctx context.Context, batchID, jobID string) error

+ MarkBatchJobSucceeded func(ctx context.Context, batchID, jobID string) (BatchState, bool, error)

+ MarkCallbackInvoked func(ctx context.Context, key string) (bool, error)

+ Prune func(ctx context.Context, before time.Time) error

v0.2.1

Jun 7, 2026

v0.2.0

Jun 7, 2026

v0.1.17

Jun 7, 2026

Changes in this version

+ func PhysicalQueueName(defaultQueue string, queueName string) string

+ func PhysicalQueueWeights(defaultQueue string, weights map[string]int) map[string]int

v0.1.16

May 6, 2026

v0.1.15

May 6, 2026

v0.1.14

May 6, 2026

Changes in this version

type Option

+ func WithHandlerContextDecorator(fn func(context.Context) context.Context) Option

v0.1.13

May 6, 2026

v0.1.12

May 6, 2026

Changes in this version

type FakeQueue

+ func (f *FakeQueue) WithContext(ctx context.Context) queueRuntime

type Queue

+ func (r *Queue) WithContext(ctx context.Context) *Queue

v0.1.11

May 6, 2026

Changes in this version

type EventKind

+ const EventDispatchFailed

+ const EventDispatchStarted

+ const EventDispatchSucceeded

v0.1.10

May 5, 2026

v0.1.9

May 5, 2026

v0.1.8

Apr 22, 2026

Changes in this version

+ func ResolveObservedJobType(rawType string, payload []byte) string

v0.1.7

Apr 6, 2026

v0.1.6

Mar 16, 2026

Changes in this version

+ var ErrBackoffUnsupported = errors.New("backoff option is not supported by this driver")

+ var ErrDuplicate = errors.New("duplicate job")

+ var ErrPauseUnsupported = errors.New("pause/resume is not supported by this driver")

+ var ErrQueueAdminUnsupported = errors.New("queue admin is not supported by this driver")

+ var ErrQueuePaused = errors.New("queue is paused")

+ var ErrQueuerShuttingDown = errors.New("queue is shutting down")

+ var ErrWorkerpoolQueueNotInitialized = errors.New("workerpool queue not initialized")

+ var ErrWorkflowNotFound = bus.ErrNotFound

+ func CancelJob(ctx context.Context, q any, jobID string) error

+ func ClearQueue(ctx context.Context, q any, queueName string) error

+ func DeleteJob(ctx context.Context, q any, queueName, jobID string) error

+ func Pause(ctx context.Context, q any, queueName string) error

+ func Ready(ctx context.Context, q any) error

+ func Resume(ctx context.Context, q any, queueName string) error

+ func RetryJob(ctx context.Context, q any, queueName, jobID string) error

+ func SafeObserve(observer Observer, event Event)

+ func SupportsNativeStats(q any) bool

+ func SupportsPause(q any) bool

+ func SupportsQueueAdmin(q any) bool

+ func ValidateDriverJob(job Job) error

+ type BatchBuilder interface

+ AllowFailures func() BatchBuilder

+ Catch func(fn func(ctx context.Context, st BatchState, err error) error) BatchBuilder

+ Dispatch func(ctx context.Context) (string, error)

+ Finally func(fn func(ctx context.Context, st BatchState) error) BatchBuilder

+ Name func(name string) BatchBuilder

+ OnQueue func(queue string) BatchBuilder

+ Progress func(fn func(ctx context.Context, st BatchState) error) BatchBuilder

+ Then func(fn func(ctx context.Context, st BatchState) error) BatchBuilder

+ type BatchState = bus.BatchState

+ type ChainBuilder interface

+ Catch func(fn func(ctx context.Context, st ChainState, err error) error) ChainBuilder

+ Dispatch func(ctx context.Context) (string, error)

+ Finally func(fn func(ctx context.Context, st ChainState) error) ChainBuilder

+ OnQueue func(queue string) ChainBuilder

+ type ChainState = bus.ChainState

+ type ChannelObserver struct

+ DropIfFull bool

+ Events chan<- Event

+ func (c ChannelObserver) Observe(event Event)

+ type Config struct

+ DefaultQueue string

+ Driver Driver

+ Logger Logger

+ Observer Observer

+ type DatabaseConfig struct

+ AutoMigrate bool

+ DB *sql.DB

+ DSN string

+ DefaultQueue string

+ DriverName string

+ Logger Logger

+ Observer Observer

+ PollInterval time.Duration

+ ProcessingLeaseNoTimeout time.Duration

+ ProcessingRecoveryGrace time.Duration

+ Workers int

+ type DispatchRecord struct

+ Job Job

+ Queue string

+ type DispatchResult = bus.DispatchResult

+ type Driver string

+ const DriverDatabase

+ const DriverNATS

+ const DriverNull

+ const DriverRabbitMQ

+ const DriverRedis

+ const DriverSQS

+ const DriverSync

+ const DriverWorkerpool

+ type DriverJobOptions struct

+ Attempt int

+ Backoff *time.Duration

+ Delay time.Duration

+ MaxRetry *int

+ QueueName string

+ Timeout *time.Duration

+ UniqueTTL time.Duration

+ func DriverOptions(job Job) DriverJobOptions

+ type Event struct

+ Attempt int

+ Driver Driver

+ Duration time.Duration

+ Err error

+ JobKey string

+ JobType string

+ Kind EventKind

+ MaxRetry int

+ Queue string

+ Scheduled bool

+ Time time.Time

+ type EventKind string

+ const EventEnqueueAccepted

+ const EventEnqueueCanceled

+ const EventEnqueueDuplicate

+ const EventEnqueueRejected

+ const EventProcessArchived

+ const EventProcessFailed

+ const EventProcessRecovered

+ const EventProcessRetried

+ const EventProcessStarted

+ const EventProcessSucceeded

+ const EventQueuePaused

+ const EventQueueResumed

+ const EventRepublishFailed

+ type FailOnError = bus.FailOnError

+ type FakeQueue struct

+ func NewFake() *FakeQueue

+ func (f *FakeQueue) AssertCount(t testing.TB, expected int)

+ func (f *FakeQueue) AssertDispatched(t testing.TB, jobType string)

+ func (f *FakeQueue) AssertDispatchedOn(t testing.TB, queueName, jobType string)

+ func (f *FakeQueue) AssertDispatchedTimes(t testing.TB, jobType string, expected int)

+ func (f *FakeQueue) AssertNotDispatched(t testing.TB, jobType string)

+ func (f *FakeQueue) AssertNothingDispatched(t testing.TB)

+ func (f *FakeQueue) BusDispatch(ctx context.Context, jobType string, payload []byte, ...) error

+ func (f *FakeQueue) BusRegister(string, busruntime.Handler)

+ func (f *FakeQueue) Dispatch(job any) error

+ func (f *FakeQueue) DispatchCtx(ctx context.Context, job any) error

+ func (f *FakeQueue) Driver() Driver

+ func (f *FakeQueue) Ready(ctx context.Context) error

+ func (f *FakeQueue) Records() []DispatchRecord

+ func (f *FakeQueue) Register(string, Handler)

+ func (f *FakeQueue) Reset()

+ func (f *FakeQueue) Shutdown(context.Context) error

+ func (f *FakeQueue) StartWorkers(context.Context) error

+ func (f *FakeQueue) Workers(int) queueRuntime

+ type Handler func(ctx context.Context, job Job) error

+ type Job struct

+ Type string

+ func DriverWithAttempt(job Job, attempt int) Job

+ func NewJob(jobType string) Job

+ func (t Job) Backoff(backoff time.Duration) Job

+ func (t Job) Bind(dst any) error

+ func (t Job) Delay(delay time.Duration) Job

+ func (t Job) OnQueue(name string) Job

+ func (t Job) Payload(payload any) Job

+ func (t Job) PayloadBytes() []byte

+ func (t Job) PayloadJSON(v any) Job

+ func (t Job) Retry(maxRetry int) Job

+ func (t Job) Timeout(timeout time.Duration) Job

+ func (t Job) UniqueFor(ttl time.Duration) Job

+ type JobSnapshot struct

+ Attempt int

+ CompletedAt *time.Time

+ ID string

+ LastError string

+ MaxRetry int

+ NextProcessAt *time.Time

+ Payload string

+ Queue string

+ State JobState

+ Type string

+ type JobState string

+ const JobStateActive

+ const JobStateArchived

+ const JobStateCompleted

+ const JobStatePending

+ const JobStateRetry

+ const JobStateScheduled

+ type ListJobsOptions struct

+ Page int

+ PageSize int

+ Queue string

+ State JobState

+ func (o ListJobsOptions) Normalize() ListJobsOptions

+ type ListJobsResult struct

+ Jobs []JobSnapshot

+ Total int64

+ func ListJobs(ctx context.Context, q any, opts ListJobsOptions) (ListJobsResult, error)

+ type Lock = bus.Lock

+ type Locker = bus.Locker

+ type Logger interface

+ Debug func(args ...interface{})

+ Error func(args ...interface{})

+ Fatal func(args ...interface{})

+ Info func(args ...interface{})

+ Warn func(args ...interface{})

+ type Message = bus.Context

+ type Middleware = bus.Middleware

+ type MiddlewareFunc = bus.MiddlewareFunc

+ type Next = bus.Next

+ type NoopLogger struct

+ func (NoopLogger) Debug(...interface{})

+ func (NoopLogger) Error(...interface{})

+ func (NoopLogger) Fatal(...interface{})

+ func (NoopLogger) Info(...interface{})

+ func (NoopLogger) Warn(...interface{})

+ type Observer interface

+ Observe func(event Event)

+ func MultiObserver(observers ...Observer) Observer

+ type ObserverFunc func(event Event)

+ func (f ObserverFunc) Observe(event Event)

+ type Option func(*runtimeOptions)

+ func WithClock(clock func() time.Time) Option

+ func WithMiddleware(middlewares ...Middleware) Option

+ func WithObserver(observer WorkflowObserver) Option

+ func WithStore(store WorkflowStore) Option

+ func WithWorkers(count int) Option

+ type Queue struct

+ func New(cfg Config, opts ...Option) (*Queue, error)

+ func NewNull(opts ...Option) (*Queue, error)

+ func NewSync(opts ...Option) (*Queue, error)

+ func NewWorkerpool(opts ...Option) (*Queue, error)

+ func (r *Queue) Batch(jobs ...Job) BatchBuilder

+ func (r *Queue) CancelJob(ctx context.Context, jobID string) error

+ func (r *Queue) Chain(jobs ...Job) ChainBuilder

+ func (r *Queue) ClearQueue(ctx context.Context, queueName string) error

+ func (r *Queue) DeleteJob(ctx context.Context, queueName, jobID string) error

+ func (r *Queue) Dispatch(job Job) (DispatchResult, error)

+ func (r *Queue) DispatchCtx(ctx context.Context, job Job) (DispatchResult, error)

+ func (r *Queue) Driver() Driver

+ func (r *Queue) FindBatch(ctx context.Context, batchID string) (BatchState, error)

+ func (r *Queue) FindChain(ctx context.Context, chainID string) (ChainState, error)

+ func (r *Queue) History(ctx context.Context, queueName string, window QueueHistoryWindow) ([]QueueHistoryPoint, error)

+ func (r *Queue) ListJobs(ctx context.Context, opts ListJobsOptions) (ListJobsResult, error)

+ func (r *Queue) Pause(ctx context.Context, queueName string) error

+ func (r *Queue) Prune(ctx context.Context, before time.Time) error

+ func (r *Queue) Ready(ctx context.Context) error

+ func (r *Queue) Register(jobType string, handler func(context.Context, Message) error)

+ func (r *Queue) Resume(ctx context.Context, queueName string) error

+ func (r *Queue) RetryJob(ctx context.Context, queueName, jobID string) error

+ func (r *Queue) Run(ctx context.Context) error

+ func (r *Queue) Shutdown(ctx context.Context) error

+ func (r *Queue) StartWorkers(ctx context.Context) error

+ func (r *Queue) Stats(ctx context.Context) (StatsSnapshot, error)

+ func (r *Queue) WithWorkers(count int) *Queue

+ type QueueAdmin interface

+ CancelJob func(ctx context.Context, jobID string) error

+ ClearQueue func(ctx context.Context, queueName string) error

+ DeleteJob func(ctx context.Context, queueName, jobID string) error

+ History func(ctx context.Context, queueName string, window QueueHistoryWindow) ([]QueueHistoryPoint, error)

+ ListJobs func(ctx context.Context, opts ListJobsOptions) (ListJobsResult, error)

+ RetryJob func(ctx context.Context, queueName, jobID string) error

+ type QueueController interface

+ Pause func(ctx context.Context, queueName string) error

+ Resume func(ctx context.Context, queueName string) error

+ type QueueCounters struct

+ Active int64

+ Archived int64

+ AvgRun time.Duration

+ AvgWait time.Duration

+ Failed int64

+ Paused int64

+ Pending int64

+ Processed int64

+ Retry int64

+ Scheduled int64

+ type QueueHistoryPoint struct

+ At time.Time

+ Failed int64

+ Processed int64

+ func QueueHistory(ctx context.Context, q any, queueName string, window QueueHistoryWindow) ([]QueueHistoryPoint, error)

+ func SinglePointHistory(snapshot StatsSnapshot, queueName string) []QueueHistoryPoint

+ func TimelineHistoryFromSnapshot(snapshot StatsSnapshot, queueName string, window QueueHistoryWindow) []QueueHistoryPoint

+ type QueueHistoryProvider interface

+ History func(ctx context.Context, queueName string, window QueueHistoryWindow) ([]QueueHistoryPoint, error)

+ type QueueHistoryWindow string

+ const QueueHistoryDay

+ const QueueHistoryHour

+ const QueueHistoryWeek

+ type QueueThroughput struct

+ Day ThroughputWindow

+ Hour ThroughputWindow

+ Week ThroughputWindow

+ type RateLimit = bus.RateLimit

+ type RateLimiter = bus.RateLimiter

+ type RetryPolicy = bus.RetryPolicy

+ type SkipWhen = bus.SkipWhen

+ type StatsCollector struct

+ func NewStatsCollector() *StatsCollector

+ func (c *StatsCollector) Observe(event Event)

+ func (c *StatsCollector) Snapshot() StatsSnapshot

+ type StatsProvider interface

+ Stats func(ctx context.Context) (StatsSnapshot, error)

+ type StatsSnapshot struct

+ ByQueue map[string]QueueCounters

+ ThroughputByQueue map[string]QueueThroughput

+ func Snapshot(ctx context.Context, q any, collector *StatsCollector) (StatsSnapshot, error)

+ func (s StatsSnapshot) Active(name string) int64

+ func (s StatsSnapshot) Archived(name string) int64

+ func (s StatsSnapshot) Failed(name string) int64

+ func (s StatsSnapshot) Paused(name string) int64

+ func (s StatsSnapshot) Pending(name string) int64

+ func (s StatsSnapshot) Processed(name string) int64

+ func (s StatsSnapshot) Queue(name string) (QueueCounters, bool)

+ func (s StatsSnapshot) Queues() []string

+ func (s StatsSnapshot) RetryCount(name string) int64

+ func (s StatsSnapshot) Scheduled(name string) int64

+ func (s StatsSnapshot) Throughput(name string) (QueueThroughput, bool)

+ type ThroughputWindow struct

+ Failed int64

+ Processed int64

+ type WithoutOverlapping = bus.WithoutOverlapping

+ type WorkerpoolConfig struct

+ DefaultJobTimeout time.Duration

+ QueueCapacity int

+ Workers int

+ type WorkflowEvent = bus.Event

+ type WorkflowEventKind = bus.EventKind

+ type WorkflowObserver = bus.Observer

+ type WorkflowObserverFunc = bus.ObserverFunc

+ type WorkflowStore = bus.Store

v0.1.5

Mar 3, 2026

v0.1.4

Mar 1, 2026

v0.1.3

Feb 27, 2026

v0.1.2

Feb 27, 2026

v0.1.1

Feb 27, 2026

v0.1.0

Feb 25, 2026

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL