queue

package module

Go to main page

Versions in this module

v0

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